serial: Expand manual tests with a fix for stalled reads

This change expands the manual test suite for the Serial API and
includes a potential fix for the problem of stalled reads on Windows.

The issue was a race between new data being received by the port and
WaitCommEvent() being called. If the data was already buffered then
the EV_RXCHAR event would not be fired. The solution is to use
ReadFile() and modify the configured timeouts so that it will wait
for at least one character to be received before returning.

Bug: 1074389
Change-Id: I4dadb5b5d5992e530a2a86161ebea87b7ae16515
Reviewed-on: https://chromium-review.googlesource.com/c/chromium/src/+/2278843
Commit-Queue: Reilly Grant <reillyg@chromium.org>
Reviewed-by: Matt Reynolds <mattreynolds@chromium.org>
Cr-Commit-Position: refs/heads/master@{#789749}

services/device/serial/serial_io_handler_win.cc

@@ -231,14 +231,12 @@ void SerialIoHandlerWin::OnDeviceRemoved(const base::string16& device_path) {
 }
 
 bool SerialIoHandlerWin::PostOpen() {
-  DCHECK(!comm_context_);
   DCHECK(!read_context_);
   DCHECK(!write_context_);
 
   base::CurrentIOThread::Get()->RegisterIOHandler(file().GetPlatformFile(),
                                                   this);
 
-  comm_context_.reset(new base::MessagePumpForIO::IOContext());
   read_context_.reset(new base::MessagePumpForIO::IOContext());
   write_context_.reset(new base::MessagePumpForIO::IOContext());
 
@@ -249,12 +247,8 @@ bool SerialIoHandlerWin::PostOpen() {
   ui_thread_task_runner()->PostTask(
       FROM_HERE, base::BindOnce(&UiThreadHelper::Start, helper_));
 
-  // A ReadIntervalTimeout of MAXDWORD will cause async reads to complete
-  // immediately with any data that's available, even if there is none.
-  // This is OK because we never issue a read request until WaitCommEvent
-  // signals that data is available.
   COMMTIMEOUTS timeouts = {0};
-  timeouts.ReadIntervalTimeout = MAXDWORD;
+  timeouts.ReadIntervalTimeout = 1;
   if (!::SetCommTimeouts(file().GetPlatformFile(), &timeouts)) {
     VPLOG(1) << "Failed to set serial timeouts";
     return false;
@@ -268,23 +262,17 @@ void SerialIoHandlerWin::ReadImpl() {
   DCHECK(pending_read_buffer());
   DCHECK(file().IsValid());
 
-  if (is_comm_pending_) {
-    // Reuse the call to WaitCommEvent() from a canceled read.
+  ClearPendingError();
+  if (!IsReadPending())
     return;
-  }
-
-  if (!SetCommMask(file().GetPlatformFile(), EV_RXCHAR)) {
-    VPLOG(1) << "Failed to set serial event flags";
-  }
 
-  event_mask_ = 0;
-  BOOL ok = ::WaitCommEvent(file().GetPlatformFile(), &event_mask_,
-                            &comm_context_->overlapped);
+  BOOL ok = ::ReadFile(file().GetPlatformFile(), pending_read_buffer(),
+                       pending_read_buffer_len(), nullptr,
+                       &read_context_->overlapped);
   if (!ok && GetLastError() != ERROR_IO_PENDING) {
-    VPLOG(1) << "Failed to receive serial event";
+    VPLOG(1) << "Read failed";
     QueueReadCompleted(0, mojom::SerialReceiveError::SYSTEM_ERROR);
   }
-  is_comm_pending_ = true;
 }
 
 void SerialIoHandlerWin::WriteImpl() {
@@ -305,15 +293,8 @@ void SerialIoHandlerWin::CancelReadImpl() {
   DCHECK_CALLED_ON_VALID_SEQUENCE(sequence_checker_);
   DCHECK(file().IsValid());
 
-  if (is_comm_pending_) {
-    // Clearing the event mask will cause an overlapped call to WaitCommEvent()
-    // to complete immediately.
-    if (!SetCommMask(file().GetPlatformFile(), 0))
-      VPLOG(1) << "Failed to clear event mask";
-  } else {
   if (!PurgeComm(file().GetPlatformFile(), PURGE_RXABORT))
     VPLOG(1) << "RX abort failed";
-  }
 }
 
 void SerialIoHandlerWin::CancelWriteImpl() {
@@ -384,50 +365,7 @@ void SerialIoHandlerWin::OnIOCompleted(
     DWORD bytes_transferred,
     DWORD error) {
   DCHECK_CALLED_ON_VALID_SEQUENCE(sequence_checker_);
-  if (context == comm_context_.get()) {
-    is_comm_pending_ = false;
-
-    DWORD errors;
-    if (!ClearCommError(file().GetPlatformFile(), &errors, nullptr)) {
-      VPLOG(1) << "Failed to clear communication error";
-      ReadCompleted(0, mojom::SerialReceiveError::SYSTEM_ERROR);
-      return;
-    }
-
-    if (errors != 0) {
-      if (errors & CE_BREAK) {
-        ReadCompleted(0, mojom::SerialReceiveError::BREAK);
-      } else if (errors & CE_FRAME) {
-        ReadCompleted(0, mojom::SerialReceiveError::FRAME_ERROR);
-      } else if (errors & CE_OVERRUN) {
-        ReadCompleted(0, mojom::SerialReceiveError::OVERRUN);
-      } else if (errors & CE_RXOVER) {
-        ReadCompleted(0, mojom::SerialReceiveError::BUFFER_OVERFLOW);
-      } else if (errors & CE_RXPARITY) {
-        ReadCompleted(0, mojom::SerialReceiveError::PARITY_ERROR);
-      } else {
-        NOTIMPLEMENTED() << "Unexpected communication error: " << std::hex
-                         << errors;
-      }
-      return;
-    }
-
-    if (read_canceled()) {
-      ReadCompleted(bytes_transferred, read_cancel_reason());
-    } else if (error != ERROR_SUCCESS && error != ERROR_OPERATION_ABORTED) {
-      VLOG(1) << "Waiting for communcations event failed: "
-              << logging::SystemErrorCodeToString(error);
-      ReadCompleted(0, mojom::SerialReceiveError::SYSTEM_ERROR);
-    } else if (pending_read_buffer()) {
-      BOOL ok = ::ReadFile(file().GetPlatformFile(), pending_read_buffer(),
-                           pending_read_buffer_len(), NULL,
-                           &read_context_->overlapped);
-      if (!ok && GetLastError() != ERROR_IO_PENDING) {
-        VPLOG(1) << "Read failed";
-        ReadCompleted(0, mojom::SerialReceiveError::SYSTEM_ERROR);
-      }
-    }
-  } else if (context == read_context_.get()) {
+  if (context == read_context_.get()) {
     if (read_canceled()) {
       ReadCompleted(bytes_transferred, read_cancel_reason());
     } else if (error == ERROR_SUCCESS || error == ERROR_OPERATION_ABORTED) {
@@ -447,10 +385,10 @@ void SerialIoHandlerWin::OnIOCompleted(
       WriteCompleted(0, mojom::SerialSendError::SYSTEM_ERROR);
       if (error == ERROR_GEN_FAILURE && IsReadPending()) {
         // For devices using drivers such as FTDI, CP2xxx, when device is
-        // disconnected, the context is comm_context_ and the error is
+        // disconnected, the context is |read_context_| and the error is
         // ERROR_OPERATION_ABORTED.
         // However, for devices using CDC-ACM driver, when device is
-        // disconnected, the context is write_context_ and the error is
+        // disconnected, the context is |write_context_| and the error is
         // ERROR_GEN_FAILURE. In this situation, in addition to a write error
         // signal, also need to generate a read error signal
         // mojom::SerialOnReceiveError which will notify the app about the
@@ -463,6 +401,31 @@ void SerialIoHandlerWin::OnIOCompleted(
   }
 }
 
+void SerialIoHandlerWin::ClearPendingError() {
+  DWORD errors;
+  if (!ClearCommError(file().GetPlatformFile(), &errors, nullptr)) {
+    VPLOG(1) << "Failed to clear communication error";
+    ReadCompleted(0, mojom::SerialReceiveError::SYSTEM_ERROR);
+    return;
+  }
+
+  if (errors & CE_BREAK) {
+    ReadCompleted(0, mojom::SerialReceiveError::BREAK);
+  } else if (errors & CE_FRAME) {
+    ReadCompleted(0, mojom::SerialReceiveError::FRAME_ERROR);
+  } else if (errors & CE_OVERRUN) {
+    ReadCompleted(0, mojom::SerialReceiveError::OVERRUN);
+  } else if (errors & CE_RXOVER) {
+    ReadCompleted(0, mojom::SerialReceiveError::BUFFER_OVERFLOW);
+  } else if (errors & CE_RXPARITY) {
+    ReadCompleted(0, mojom::SerialReceiveError::PARITY_ERROR);
+  } else if (errors != 0) {
+    NOTIMPLEMENTED() << "Unexpected communication error: " << std::hex
+                     << errors;
+    ReadCompleted(0, mojom::SerialReceiveError::SYSTEM_ERROR);
+  }
+}
+
 void SerialIoHandlerWin::Flush(mojom::SerialPortFlushMode mode) const {
   DWORD flags;
   switch (mode) {

services/device/serial/serial_io_handler_win.h

@@ -48,25 +48,15 @@ class SerialIoHandlerWin : public SerialIoHandler,
                      DWORD bytes_transfered,
                      DWORD error) override;
 
+  void ClearPendingError();
   void OnDeviceRemoved(const base::string16& device_path);
 
-  // Context used for asynchronous WaitCommEvent calls.
-  std::unique_ptr<base::MessagePumpForIO::IOContext> comm_context_;
-
   // Context used for overlapped reads.
   std::unique_ptr<base::MessagePumpForIO::IOContext> read_context_;
 
   // Context used for overlapped writes.
   std::unique_ptr<base::MessagePumpForIO::IOContext> write_context_;
 
-  // Asynchronous event mask state
-  DWORD event_mask_ = 0;
-
-  // Indicates if a pending read is waiting on initial data arrival via
-  // WaitCommEvent, as opposed to waiting on actual ReadFile completion
-  // after a corresponding WaitCommEvent has completed.
-  bool is_comm_pending_ = false;
-
   // The helper lives on the UI thread and holds a weak reference back to the
   // handler that owns it.
   UiThreadHelper* helper_ = nullptr;

third_party/blink/web_tests/wpt_internal/serial/serialPort_loopback-manual.https.html

@@ -15,55 +15,167 @@
     </p>
     <script>
       manual_loopback_serial_test(async (t, port) => {
-        await port.open({baudrate: 9600, buffersize: 64});
+        await port.open({baudrate: 115200, buffersize: 1024});
 
-        const data = new Uint8Array(1024);  // Much larger than buffersize above.
+        // Create something much smaller than buffersize above.
+        const data = new Uint8Array(64);
         for (let i = 0; i < data.byteLength; ++i)
           data[i] = i & 0xff;
 
+        const reader = port.readable.getReader();
+
+        for (let i = 0; i < 10; ++i) {
           const writer = port.writable.getWriter();
           writer.write(data);
           const writePromise = writer.close();
 
-        const reader = port.readable.getReader();
           const value = await readWithLength(reader, data.byteLength);
+          await writePromise;
+
+          compareArrays(value, data);
+        }
+
         reader.releaseLock();
+        await port.close();
+      }, 'Can perform a series of small writes.');
+
+      manual_loopback_serial_test(async (t, port) => {
+        await port.open({baudrate: 115200, buffersize: 1024});
+
+        // Create something much larger than buffersize above.
+        const data = new Uint8Array(10 * 1024);
+        for (let i = 0; i < data.byteLength; ++i)
+          data[i] = (i / 1024) & 0xff;
+
+        const reader = port.readable.getReader();
+
+        for (let i = 0; i < 10; ++i) {
+          const writer = port.writable.getWriter();
+          writer.write(data);
+          const writePromise = writer.close();
 
+          const value = await readWithLength(reader, data.byteLength);
           await writePromise;
 
           compareArrays(value, data);
+        }
 
+        reader.releaseLock();
         await port.close();
-      }, 'Can perform a large write.');
+      }, 'Can perform a series of large writes.');
 
       manual_loopback_serial_test(async (t, port) => {
         await port.open({baudrate: 115200, buffersize: 64});
 
-        let data = new Uint8Array([1, 2, 3, 4, 5, 6, 7, 8]);
-
-        let writer = port.writable.getWriter();
-        writer.write(data);
+        const writer = port.writable.getWriter();
         // |data| is small enough to be completely transmitted.
-        await writer.close();
+        let data = new Uint8Array([1, 2, 3, 4, 5, 6, 7, 8]);
+        await writer.write(data);
 
         // Wait a little bit for the device to process the incoming data.
         await new Promise((resolve) => setTimeout(resolve, 100));
-
+        // ...before discarding the receive buffers.
         await port.readable.cancel();
 
         data = new Uint8Array([9, 10, 11, 12, 13, 14, 15, 16]);
-        writer = port.writable.getWriter();
+        const reader = port.readable.getReader();
+        const readPromise = readWithLength(reader, data.byteLength);
+
+        // The next block of data should be received successfully.
+        await writer.write(data);
+        writer.releaseLock();
+
+        const value = await readPromise;
+        reader.releaseLock();
+
+        compareArrays(value, data);
+
+        await port.close();
+      }, 'Canceling the reader discards buffered data.');
+
+      manual_loopback_serial_test(async (t, port) => {
+        await port.open({baudrate: 115200, buffersize: 1024});
+
+        // Create something much larger than buffersize above.
+        const data = new Uint8Array(16 * 1024);
+        for (let i = 0; i < data.byteLength; ++i)
+          data[i] = (i / 1024) & 0xff;
+
+        // Completely write |data| to the port without waiting for it to be
+        // received.
+        const writer = port.writable.getWriter();
         writer.write(data);
         await writer.close();
 
         const reader = port.readable.getReader();
-        const value = await readWithLength(reader, data.byteLength);
+        const chunks = [];
+        let actualLength = 0;
+        while (true) {
+          try {
+            const {value, done} = await reader.read();
+            if (value) {
+              actualLength += value.byteLength;
+              chunks.push(value);
+            }
+            if (done) {
+              assert_unreached("Unexpected end of stream.");
+              break;
+            }
+          } catch (e) {
+            assert_equals(e.name, 'BufferOverrunError');
+            break;
+          }
+        }
         reader.releaseLock();
 
-        compareArrays(value, data);
+        const buffer = new Uint8Array(actualLength);
+        chunks.reduce((offset, chunk) => {
+          buffer.set(chunk, offset);
+          return offset + chunk.byteLength;
+        }, 0);
+
+        assert_greater_than(actualLength, 0);
+        compareArrays(buffer, data.slice(0, actualLength));
 
         await port.close();
-      }, 'Canceling the reader discards buffered data.');
+      }, 'Overflowing the receive buffer triggers an error.');
+
+      manual_loopback_serial_test(async (t, port) => {
+        await port.open({baudrate: 115200, buffersize: 1024});
+
+        let reader = port.readable.getReader();
+        let readPromise = (async () => {
+          // A single zero byte will be read before the break is detected.
+          const {value, done} = await reader.read();
+          compareArrays(value, new Uint8Array([0]));
+          assert_false(done);
+
+          try {
+            const {value, done} = await reader.read();
+            assert_unreached(`Expected break, got ${value.byteLength} bytes`);
+          } catch (e) {
+            assert_equals(e.constructor, DOMException);
+            assert_equals(e.name, 'BreakError');
+          }
+        })();
+
+        await port.setSignals({brk: true});
+        await readPromise;
+        await port.setSignals({brk: false});
+
+        const writer = port.writable.getWriter();
+        // |data| is small enough to be completely transmitted.
+        let data = new Uint8Array([1, 2, 3, 4, 5, 6, 7, 8]);
+        await writer.write(data);
+        writer.releaseLock();
+
+        reader = port.readable.getReader();
+        const buffer = await readWithLength(reader, data.byteLength);;
+        compareArrays(buffer, data);
+        reader.releaseLock();
+
+        await port.close();
+      }, 'Break is detected.');
     </script>
   </body>
 </html>