IO unset read/write (#3)

* Update gosec build step

* IO can unset reads/writes

* Close() unsets reads+writes and deregisters the IO slot

* AsyncAdapter exposes its IO Slot

.github/workflows/gosec.yml

@@ -15,10 +15,10 @@ jobs:
       - name: Checkout Source
         uses: actions/checkout@v3
       - name: Run Gosec Security Scanner
-        uses: securego/gosec@v2.15.0
+        uses: securego/gosec@v2.21.0
         with:
           args: '-no-fail -fmt=sarif -out=results.sarif -exclude-dir=examples -exclude-dir=stress_test -exclude-dir=other -exclude-dir=docs -exclude-dir=tests -exclude-dir=benchmark ./...'
       - name: Upload SARIF file
-        uses: github/codeql-action/upload-sarif@v2
+        uses: github/codeql-action/upload-sarif@v3
         with:
           sarif_file: results.sarif

async_adapter.go

@@ -202,7 +202,8 @@ func (a *AsyncAdapter) Close() error {
 		return io.EOF
 	}
 
-	_ = a.ioc.poller.Del(&a.slot)
+	_ = a.ioc.UnsetReadWrite(&a.slot)
+	a.ioc.Deregister(&a.slot)
 
 	return syscall.Close(a.slot.Fd)
 }
@@ -245,3 +246,7 @@ func (a *AsyncAdapter) cancelWrites() {
 func (a *AsyncAdapter) RawFd() int {
 	return a.slot.Fd
 }
+
+func (a *AsyncAdapter) Slot() *internal.Slot {
+	return &a.slot
+}

file.go

@@ -232,10 +232,10 @@ func (f *file) Close() error {
 		return io.EOF
 	}
 
-	err := f.ioc.poller.Del(&f.slot)
-	if err != nil {
+	if err := f.ioc.UnsetReadWrite(&f.slot); err != nil {
 		return err
 	}
+	f.ioc.Deregister(&f.slot)
 
 	return syscall.Close(f.slot.Fd)
 }

io.go

@@ -78,14 +78,39 @@ func (ioc *IO) Deregister(slot *internal.Slot) {
 	}
 }
 
+// SetRead tells the kernel to notify us when reads can be made on the provided IO slot. If successful, this call must
+// be succeeded by Register(slot).
+//
+// It is safe to call this method multiple times.
 func (ioc *IO) SetRead(slot *internal.Slot) error {
 	return ioc.poller.SetRead(slot)
 }
 
+// UnsetRead tells the kernel to not notify us anymore when reads can be made on the provided IO slot. Since the
+// underlying platform-specific poller already unsets a read before dispatching it, callers must only use this method
+// they want to cancel a currently-scheduled read. For example, when an error occurs outside of an AsyncRead call and
+// the underlying file descriptor must be closed. In that case, this call must be succeeded by Deregister(slot).
+//
+// It is safe to call this method multiple times.
+func (ioc *IO) UnsetRead(slot *internal.Slot) error {
+	return ioc.poller.DelRead(slot)
+}
+
+// Like SetRead but for writes.
 func (ioc *IO) SetWrite(slot *internal.Slot) error {
 	return ioc.poller.SetWrite(slot)
 }
 
+// Like UnsetRead but for writes.
+func (ioc *IO) UnsetWrite(slot *internal.Slot) error {
+	return ioc.poller.DelWrite(slot)
+}
+
+// UnsetRead and UnsetWrite in a single call.
+func (ioc *IO) UnsetReadWrite(slot *internal.Slot) error {
+	return ioc.poller.Del(slot)
+}
+
 // Run runs the event processing loop.
 func (ioc *IO) Run() error {
 	for {

io_test.go

@@ -316,6 +316,38 @@ func TestIOPending(t *testing.T) {
 	}
 }
 
+func TestSetUnsetRead(t *testing.T) {
+	ioc := MustIO()
+	defer ioc.Close()
+
+	pipe, err := internal.NewPipe()
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	if pipe.ReadFd() != pipe.Slot().Fd {
+		t.Fatal("pipe must be identified by its read end file descriptor")
+	}
+
+	for i := 0; i < 10; i++ {
+		if err := ioc.SetRead(pipe.Slot()); err != nil {
+			t.Fatal(err)
+		}
+	}
+
+	for i := 0; i < 100; i++ {
+		if err := ioc.UnsetRead(pipe.Slot()); err != nil {
+			t.Fatal(err)
+		}
+	}
+
+	for i := 0; i < 100; i++ {
+		if err := ioc.UnsetReadWrite(pipe.Slot()); err != nil {
+			t.Fatal(err)
+		}
+	}
+}
+
 func BenchmarkPollOne(b *testing.B) {
 	ioc := MustIO()
 	defer ioc.Close()

listen_conn.go

@@ -113,7 +113,8 @@ func (l *listener) accept() (Conn, error) {
 }
 
 func (l *listener) Close() error {
-	_ = l.ioc.poller.Del(&l.slot)
+	_ = l.ioc.UnsetReadWrite(&l.slot)
+	l.ioc.Deregister(&l.slot)
 	return syscall.Close(l.slot.Fd)
 }
 

multicast/peer.go

@@ -676,6 +676,8 @@ func (p *UDPPeer) LocalAddr() *net.UDPAddr {
 func (p *UDPPeer) Close() error {
 	if !p.closed {
 		p.closed = true
+		_ = p.ioc.UnsetReadWrite(&p.slot)
+		p.ioc.Deregister(&p.slot)
 		return p.socket.Close()
 	}
 	return nil

packet.go

@@ -195,6 +195,8 @@ func (c *packetConn) getWriteHandler(b []byte, to net.Addr, cb AsyncWriteCallbac
 
 func (c *packetConn) Close() error {
 	atomic.StoreUint32(&c.closed, 1)
+	_ = c.ioc.UnsetReadWrite(&c.slot)
+	c.ioc.Deregister(&c.slot)
 	return syscall.Close(c.slot.Fd)
 }