Handle EPOLLHUP in EpollSystem

core/native/src/main/scala/cats/effect/unsafe/EpollSystem.scala

@@ -105,7 +105,7 @@ object EpollSystem extends PollingSystem {
     private[this] var writeCallback: Either[Throwable, Int] => Unit = null
 
     def notify(events: Int): Unit = {
-      if ((events & EPOLLIN) != 0) {
+      if ((events & EPOLLIN) != 0 || (events & EPOLLHUP) != 0) {
         val counter = readReadyCounter + 1
         readReadyCounter = counter
         val cb = readCallback
@@ -269,6 +269,7 @@ object EpollSystem extends PollingSystem {
     final val EPOLLOUT = 0x004
     final val EPOLLONESHOT = 1 << 30
     final val EPOLLET = 1 << 31
+    final val EPOLLHUP = 1 << 4
 
     type epoll_event
     type epoll_data_t = Ptr[Byte]

tests/native/src/test/scala/cats/effect/FileDescriptorPollerSpec.scala

@@ -51,21 +51,21 @@ class FileDescriptorPollerSpec extends BaseSpec {
           IO(guard(unistd.write(writeFd, buf.atUnsafe(offset), length.toULong)))
         }
         .void
+  }
 
-    private def guard(thunk: => CInt): Either[Unit, CInt] = {
-      val rtn = thunk
-      if (rtn < 0) {
-        val en = errno
-        if (en == EAGAIN || en == EWOULDBLOCK)
-          Left(())
-        else
-          throw new IOException(fromCString(strerror(errno)))
-      } else
-        Right(rtn)
-    }
+  def guard(thunk: => CInt): Either[Unit, CInt] = {
+    val rtn = thunk
+    if (rtn < 0) {
+      val en = errno
+      if (en == EAGAIN || en == EWOULDBLOCK)
+        Left(())
+      else
+        throw new IOException(fromCString(strerror(errno)))
+    } else
+      Right(rtn)
   }
 
-  def mkPipe: Resource[IO, Pipe] =
+  def pipeHandle: Resource[IO, (Int, Int)] =
     Resource
       .make {
         IO {
@@ -91,15 +91,27 @@ class FileDescriptorPollerSpec extends BaseSpec {
               throw new IOException(fromCString(strerror(errno)))
           }
       }
-      .flatMap {
-        case (readFd, writeFd) =>
-          Resource.eval(FileDescriptorPoller.get).flatMap { poller =>
-            (
-              poller.registerFileDescriptor(readFd, true, false),
-              poller.registerFileDescriptor(writeFd, false, true)
-            ).mapN(new Pipe(readFd, writeFd, _, _))
+
+  def mkPipe: Resource[IO, Pipe] =
+    pipeHandle.flatMap {
+      case (readFd, writeFd) =>
+        Resource.eval(FileDescriptorPoller.get).flatMap { poller =>
+          (
+            poller.registerFileDescriptor(readFd, true, false),
+            poller.registerFileDescriptor(writeFd, false, true)
+          ).mapN(new Pipe(readFd, writeFd, _, _))
+        }
+    }
+
+  def mkReadOnlyPipe: Resource[IO, (Int, Int, FileDescriptorPollHandle)] =
+    pipeHandle.flatMap {
+      case (readFd, writeFd) =>
+        Resource.eval(FileDescriptorPoller.get).flatMap { poller =>
+          poller.registerFileDescriptor(readFd, true, false).map { readHandle =>
+            (readFd, writeFd, readHandle)
           }
-      }
+        }
+    }
 
   "FileDescriptorPoller" should {
 
@@ -140,6 +152,33 @@ class FileDescriptorPollerSpec extends BaseSpec {
         pipe.read(new Array[Byte](1), 0, 1).as(false).timeoutTo(1.second, IO.pure(true))
       }
     }
-  }
 
+    "pollReadRec reads until EPOLLHUP then terminates with EOF" in real {
+      mkReadOnlyPipe.use {
+        case (readFd, writeFd, readHandle) =>
+          for {
+            buf <- IO(new Array[Byte](1))
+            gate <- Deferred[IO, Unit]
+            reader <- {
+              readHandle.pollReadRec(()) { _ =>
+                IO(guard(unistd.read(readFd, buf.atUnsafe(0), 1.toULong))).flatTap {
+                  case Left(_) => gate.complete(()).attempt.void
+                  case Right(_) => IO.unit
+                }
+              }
+            }.start
+
+            _ <- gate.get
+
+            writer <- IO {
+              unistd.close(writeFd)
+            }.start
+
+            readerResult <- reader.joinWithNever
+            _ <- writer.joinWithNever
+          } yield readerResult must be_==(0)
+      }
+    }
+
+  }
 }