diff --git a/Source/Core/Core/ConfigManager.cpp b/Source/Core/Core/ConfigManager.cpp
index 0005a280bd..d8a0b6b52e 100644
--- a/Source/Core/Core/ConfigManager.cpp
+++ b/Source/Core/Core/ConfigManager.cpp
@@ -222,6 +222,7 @@ void SConfig::SaveCoreSettings(IniFile& ini)
   core->Set("AudioStretchMaxLatency", m_audio_stretch_max_latency);
   core->Set("AgpCartAPath", m_strGbaCartA);
   core->Set("AgpCartBPath", m_strGbaCartB);
+  core->Set("BlockingPipes", m_blockingPipes);
   core->Set("SlotA", m_EXIDevice[0]);
   core->Set("SlotB", m_EXIDevice[1]);
   core->Set("SerialPort1", m_EXIDevice[2]);
@@ -492,6 +493,7 @@ void SConfig::LoadCoreSettings(IniFile& ini)
   core->Get("AudioStretchMaxLatency", &m_audio_stretch_max_latency, 80);
   core->Get("AgpCartAPath", &m_strGbaCartA);
   core->Get("AgpCartBPath", &m_strGbaCartB);
+  core->Get("BlockingPipes", &m_blockingPipes, false);
   core->Get("SlotA", (int*)&m_EXIDevice[0], ExpansionInterface::EXIDEVICE_NONE);
   core->Get("SlotB", (int*)&m_EXIDevice[1], ExpansionInterface::EXIDEVICE_SLIPPI);
   core->Get("SerialPort1", (int*)&m_EXIDevice[2], ExpansionInterface::EXIDEVICE_NONE);
diff --git a/Source/Core/Core/HW/EXI/EXI_DeviceSlippi.cpp b/Source/Core/Core/HW/EXI/EXI_DeviceSlippi.cpp
index 53b646b188..4bb56c0677 100644
--- a/Source/Core/Core/HW/EXI/EXI_DeviceSlippi.cpp
+++ b/Source/Core/Core/HW/EXI/EXI_DeviceSlippi.cpp
@@ -38,6 +38,7 @@
 //#define CREATE_DIFF_FILES
 extern std::unique_ptr<SlippiPlaybackStatus> g_playbackStatus;
 extern std::unique_ptr<SlippiReplayComm> g_replayComm;
+extern bool g_needInputForFrame;
 
 #ifdef LOCAL_TESTING
 bool isLocalConnected = false;
@@ -2146,7 +2147,7 @@ void CEXISlippi::DMAWrite(u32 _uAddr, u32 _uSize)
     configureCommands(&memPtr[1], receiveCommandsLen);
     writeToFileAsync(&memPtr[0], receiveCommandsLen + 1, "create");
     bufLoc += receiveCommandsLen + 1;
-    // g_needInputForFrame = true;
+    g_needInputForFrame = true;
     SlippiSpectateServer::getInstance().startGame();
     SlippiSpectateServer::getInstance().write(&memPtr[0], receiveCommandsLen + 1);
   }
@@ -2154,7 +2155,7 @@ void CEXISlippi::DMAWrite(u32 _uAddr, u32 _uSize)
   if (byte == CMD_MENU_FRAME)
   {
     SlippiSpectateServer::getInstance().write(&memPtr[0], _uSize);
-    // g_needInputForFrame = true;
+    g_needInputForFrame = true;
   }
 
   INFO_LOG(EXPANSIONINTERFACE, "EXI SLIPPI DMAWrite: addr: 0x%08x size: %d, bufLoc:[%02x %02x %02x %02x %02x]",
@@ -2187,7 +2188,7 @@ void CEXISlippi::DMAWrite(u32 _uAddr, u32 _uSize)
       prepareFrameData(&memPtr[bufLoc + 1]);
       break;
     case CMD_FRAME_BOOKEND:
-      // g_needInputForFrame = true;
+      g_needInputForFrame = true;
       writeToFileAsync(&memPtr[bufLoc], payloadLen + 1, "");
       SlippiSpectateServer::getInstance().write(&memPtr[bufLoc], payloadLen + 1);
       break;
diff --git a/Source/Core/Core/Slippi/SlippiNetplay.cpp b/Source/Core/Core/Slippi/SlippiNetplay.cpp
index 952dc7972f..64a5f9a0f9 100644
--- a/Source/Core/Core/Slippi/SlippiNetplay.cpp
+++ b/Source/Core/Core/Slippi/SlippiNetplay.cpp
@@ -182,6 +182,13 @@ unsigned int SlippiNetplayClient::OnData(sf::Packet& packet)
       int32_t headFrame = remotePadQueue.empty() ? 0 : remotePadQueue.front()->frame;
       int inputsToCopy = frame - headFrame;
 
+      // Check that the packet actually contains the data it claims to
+      //if ((5 + inputsToCopy * SLIPPI_PAD_DATA_SIZE) > (int)packet.getDataSize())
+      //{
+      //  ERROR_LOG(SLIPPI_ONLINE, "Netplay packet too small to read pad buffer");
+      //  break;
+      //}
+
       for (int i = inputsToCopy - 1; i >= 0; i--)
       {
         auto pad = std::make_unique<SlippiPad>(frame - i, &packetData[5 + i * SLIPPI_PAD_DATA_SIZE]);
diff --git a/Source/Core/InputCommon/ControllerInterface/Pipes/Pipes.cpp b/Source/Core/InputCommon/ControllerInterface/Pipes/Pipes.cpp
index 8a3982145b..25c413984f 100644
--- a/Source/Core/InputCommon/ControllerInterface/Pipes/Pipes.cpp
+++ b/Source/Core/InputCommon/ControllerInterface/Pipes/Pipes.cpp
@@ -14,11 +14,14 @@
 #include <sys/stat.h>
 #include <vector>
 
+#include "Core/ConfigManager.h"
 #include "Common/FileUtil.h"
 #include "Common/StringUtil.h"
 #include "InputCommon/ControllerInterface/ControllerInterface.h"
 #include "InputCommon/ControllerInterface/Pipes/Pipes.h"
 
+bool g_needInputForFrame;
+
 namespace ciface::Pipes
 {
 static const std::array<std::string, 12> s_button_tokens{
@@ -151,6 +154,17 @@ s32 PipeDevice::readFromPipe(PIPE_FD file_descriptor, char* in_buffer, size_t si
     }
   }
   return (s32)bytesread;
+#ifndef _WIN32
+  if (SConfig::GetInstance().m_blockingPipes)
+  {
+    fd_set set;
+    FD_ZERO(&set);
+    FD_SET(file_descriptor, &set);
+
+    // Wait for activity on the socket
+    select(1, &set, NULL, NULL, NULL);
+  }
+#endif
 #else
   return read(file_descriptor, in_buffer, size);
 #endif
@@ -158,27 +172,30 @@ s32 PipeDevice::readFromPipe(PIPE_FD file_descriptor, char* in_buffer, size_t si
 
 void PipeDevice::UpdateInput()
 {
-  // Read any pending characters off the pipe. If we hit a newline,
-  // then dequeue a command off the front of m_buf and parse it.
-  char buf[32];
-  s32 bytes_read = readFromPipe(m_fd, buf, sizeof buf);
-  if (bytes_read == 0) {
-    // Pipe died, so just quit out
-    return;
-  }
-  while (bytes_read > 0)
-  {
-    m_buf.append(buf, bytes_read);
-    bytes_read = readFromPipe(m_fd, buf, sizeof buf);
-  }
-  std::size_t newline = m_buf.find("\n");
-  while (newline != std::string::npos)
-  {
-    std::string command = m_buf.substr(0, newline);
-    ParseCommand(command);
-    m_buf.erase(0, newline + 1);
-    newline = m_buf.find("\n");
-  }
+  bool finished = false;
+  do {
+    // Read any pending characters off the pipe. If we hit a newline,
+    // then dequeue a command off the front of m_buf and parse it.
+    char buf[32];
+    s32 bytes_read = readFromPipe(m_fd, buf, sizeof buf);
+    if (bytes_read == 0) {
+      // Pipe died, so just quit out
+      return;
+    }
+    while (bytes_read > 0)
+    {
+      m_buf.append(buf, bytes_read);
+      bytes_read = readFromPipe(m_fd, buf, sizeof buf);
+    }
+    std::size_t newline = m_buf.find("\n");
+    while (newline != std::string::npos)
+    {
+      std::string command = m_buf.substr(0, newline);
+      ParseCommand(command);
+      m_buf.erase(0, newline + 1);
+      newline = m_buf.find("\n");
+    }
+  } while (!finished && SConfig::GetInstance().m_blockingPipes);
 }
 
 void PipeDevice::AddAxis(const std::string& name, double value)
@@ -206,11 +223,16 @@ void PipeDevice::SetAxis(const std::string& entry, double value)
     search_lo->second->SetState(lo);
 }
 
-void PipeDevice::ParseCommand(const std::string& command)
+bool PipeDevice::ParseCommand(const std::string& command)
 {
   const std::vector<std::string> tokens = SplitString(command, ' ');
   if (tokens.size() < 2 || tokens.size() > 4)
-    return;
+    return false;
+  if (tokens[0] == "FLUSH")
+  {
+    g_needInputForFrame = false;
+    return true;
+  }
   if (tokens[0] == "PRESS" || tokens[0] == "RELEASE")
   {
     auto search = m_buttons.find(tokens[1]);
@@ -232,5 +254,6 @@ void PipeDevice::ParseCommand(const std::string& command)
       SetAxis(tokens[1] + " Y", y);
     }
   }
+  return false;
 }
 }  // namespace ciface::Pipes
diff --git a/Source/Core/InputCommon/ControllerInterface/Pipes/Pipes.h b/Source/Core/InputCommon/ControllerInterface/Pipes/Pipes.h
index 245eaba51c..e842aa234f 100644
--- a/Source/Core/InputCommon/ControllerInterface/Pipes/Pipes.h
+++ b/Source/Core/InputCommon/ControllerInterface/Pipes/Pipes.h
@@ -57,7 +57,7 @@ private:
   };
 
   void AddAxis(const std::string& name, double value);
-  void ParseCommand(const std::string& command);
+  bool ParseCommand(const std::string& command);
   void SetAxis(const std::string& entry, double value);
   s32 readFromPipe(PIPE_FD file_descriptor, char* in_buffer, size_t size);