Merge pull request #3603 from mtomko/backport

Backport 3591

Author: mpilquist

io/jvm/src/main/scala/fs2/io/net/SelectingSocketGroup.scala

@@ -22,6 +22,8 @@
 package fs2
 package io.net
 
+import scala.concurrent.duration._
+
 import cats.effect.LiftIO
 import cats.effect.Selector
 import cats.effect.kernel.Async
@@ -100,7 +102,11 @@ private final class SelectingSocketGroup[F[_]: LiftIO: Dns](selector: Selector)(
   ): Resource[F, (SocketAddress[IpAddress], Stream[F, Socket[F]])] =
     Resource
       .make(F.delay(selector.provider.openServerSocketChannel())) { ch =>
-        F.delay(ch.close())
+        def waitForDeregistration: F[Unit] =
+          // sleep time set to be short enough to not noticeably delay shutdown but long enough to
+          // give the runtime/cpu time to do something else; some guesswork involved here
+          F.delay(ch.isRegistered()).ifM(F.sleep(2.millis) >> waitForDeregistration, F.unit)
+        F.delay(ch.close()) >> waitForDeregistration
       }
       .evalMap { serverCh =>
         val configure = address.traverse(_.resolve).flatMap { ip =>

io/shared/src/test/scala/fs2/io/net/tcp/SocketSuite.scala

@@ -269,6 +269,18 @@ class SocketSuite extends Fs2Suite with SocketSuitePlatform {
       }
     }
 
+    test("sockets are released at the end of the resource scope") {
+      val f =
+        Network[IO].serverResource(port = Some(port"9071")).use { case (bindAddress, clients) =>
+          clients.foreach(_ => IO.sleep(1.second)).compile.drain.background.surround {
+            Network[IO].client(bindAddress).use { client =>
+              client.read(1).assertEquals(None)
+            }
+          }
+        }
+      f >> f >> f
+    }
+
     test("endOfOutput / endOfInput ignores ENOTCONN") {
       Network[IO].serverResource().use { case (bindAddress, clients) =>
         Network[IO].client(bindAddress).surround(IO.sleep(100.millis)).background.surround {