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

@@ -101,6 +101,41 @@ class FileDescriptorPollerSpec extends BaseSpec {
           }
       }
 
+  def mkReadOnlyPipe: Resource[IO, (Int, Int, FileDescriptorPollHandle)] =
+    Resource
+      .make {
+        IO {
+          val fd = stackalloc[CInt](2.toULong)
+          if (unistd.pipe(fd) != 0)
+            throw new IOException(fromCString(strerror(errno)))
+          (fd(0), fd(1))
+        }
+      } {
+        case (readFd, writeFd) =>
+          IO {
+            unistd.close(readFd)
+            unistd.close(writeFd)
+            ()
+          }
+      }
+      .evalTap {
+        case (readFd, writeFd) =>
+          IO {
+            if (fcntl(readFd, F_SETFL, O_NONBLOCK) != 0)
+              throw new IOException(fromCString(strerror(errno)))
+            if (fcntl(writeFd, F_SETFL, O_NONBLOCK) != 0)
+              throw new IOException(fromCString(strerror(errno)))
+          }
+      }
+      .flatMap {
+        case (readFd, writeFd) =>
+          Resource.eval(FileDescriptorPoller.get).flatMap { poller =>
+            poller.registerFileDescriptor(readFd, true, false).map { readHandle =>
+              (readFd, writeFd, readHandle)
+            }
+          }
+      }
+
   "FileDescriptorPoller" should {
 
     "notify read-ready events" in real {
@@ -140,6 +175,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))
+            reader <- {
+              readHandle.pollReadRec(()) { _ =>
+                IO {
+                  val n = unistd.read(readFd, buf.atUnsafe(0), 1.toULong)
+                  if (n > 0) Right(Some(n.toInt))
+                  else if (n == 0) Right(None)
+                  else if (errno == EAGAIN || errno == EWOULDBLOCK) Left(())
+                  else throw new IOException(fromCString(strerror(errno)))
+                }
+              }
+            }.start
+
+            writer <- IO {
+              unistd.close(writeFd)
+            }.start
+
+            readerResult <- reader.joinWithNever
+            _ <- writer.joinWithNever
+          } yield readerResult must be_==(None)
+      }
+    }
+
+  }
 }