[API Proposal]: Add Unix-specific APIs to register for IO ready notifications on the thread pool

[kouvel]

Background and motivation

Currently we don't have a common way of handling IO readiness and completions across operating systems. On Windows the IO completion thread pool is not very efficient. The mechanism used on Linux in System.Net.Sockets is more efficient, but comes with other issues. As part of adding IO readiness/completion handling to the portable thread pool (transitioning away from the native thread pool), we would also like to consolidate the mechanisms used for IO readiness/completion handling across operating systems. Hopefully this would ease maintenance, make it easier for other components to also use the functionality, and enable further experimentation for improving things across the various async IO mechanisms, such as potentially with better thread management and better ordering of work.

The APIs being proposed here are specific to epoll/kqueue, analogous to the existing APIs for overlapped IO on Windows. Ideally, we would move to io_uring on Linux, but it's not available in all distros yet and we would continue to need a kqueue-based solution. When we decide to add support for io_uring, it would likely require a separate set of APIs.

More discussion here: #47631

API Proposal

Update: It's not necessary to expose these APIs in contracts, so the current proposal is to make them public for use in .NET libraries but not expose them in contracts for public usage.

namespace System.Threading
{
    public static class ThreadPool
    {
        [UnsupportedOSPlatform("windows")]
        [UnsupportedOSPlatform("browser")]
        public static void RegisterForIOReadyNotifications(
            SafeHandle handle,
            IOReadyEvents events,
            IIOReadyEventHandler eventHandler) { }

        [UnsupportedOSPlatform("windows")]
        [UnsupportedOSPlatform("browser")]
        public static void UnregisterForIOReadyNotifications(SafeHandle handle) { }
    }


    [Flags]
    public enum IOReadyEvents
    {
        None = 0x0,
        Read = 0x1,
        Write = 0x2,
        CloseOrError = 0x4
    }

    public interface IIOReadyEventHandler
    {
        IOReadyEvents HandleEventsInlineIfNecessary(IOReadyEvents events);
        void HandleEvents(IOReadyEvents events);
    }
}

RegisterForIOReadyNotifications

• Registers an IO handle to receive notifications of IO-ready events. The event handler is notified of IO-ready events for the IO handle for inline processing, or asynchronous processing on thread pool worker threads.
• This API is specific to Unix-like platforms and uses epoll or kqueue. The underlying implementation may not support some types of IO handles.
• Arguments
  • handle - The IO handle to register for IO-ready event notifications. The safe handle wraps a file descriptor that typically represents a non-blocking pipe or socket. An IO handle is keyed by the handle value and may be registered only once.
  • events - A mask of events for which to register for notifications
  • eventHandler - An object that would handle IO-ready events for the IO handle
• Exceptions
  • ArgumentNullException - One of the arguments is null
  • ArgumentException - handle is closed or invalid
  • ArgumentOutOfRangeException - The value of events is invalid
  • InvalidOperationException - handle was already registered
  • IOException - An error occurred when attempting to register the IO handle for notifications. The HResult property would contain an error code for further investigation

UnregisterForIOReadyNotifications

• Unregisters an IO handle to no longer receive notifications of new IO-ready events. Any IO-ready events that were already received prior to unregisteration are still delivered to the previously registered event handler.
• Arguments
  • handle - The IO handle to unregister for new IO-ready event notifications
• Exceptions
  • ArgumentNullException - One of the arguments is null
  • InvalidOperationException - handle was not registered

IOReadyEvents

• Read / Write - The stream is ready for a read or write respectively
• CloseOrError
  • There appear to be some inconsistencies in how epoll reports the RDHUP, HUP, and ERR events on Linux, see Add test to list OS event details tokio-rs/mio#1091 (comment). To simplify things, for epoll when any of those are set, we could just include this bit and have the user determine what happened based on the error code from the following IO operation.
  • For kqueue, similarly this bit would be set when EV_EOF or EV_ERROR are reported
  • System.Net.Sockets could continue to do what it's doing now and treat CloseOrError as Read | Write

IIOReadyEventHandler

• IOReadyEvents HandleEventsInlineIfNecessary(IOReadyEvents events);

  • Currently on Unix, when a synchronous operation is performed on a non-blocking socket, after the operation completes the blocking thread is released immediately upon receiving the event instead of queuing event processing to the thread pool, since otherwise the blocking may have occurred on a thread pool thread and lead to thread starvation
  • There is currently a config option to force processing events on the polling threads inline (along with configuring the number of polling threads), which we are currently planning to keep intact, at least for now
  • This method is called immediately upon receiving an event for inline processing
  • events specifies which IO-ready events were triggered
  • The returned events, if not None, will be queued for processing and will be sent later through HandleEvents. So, a user may choose to process all or some of the events inline depending on whether there is a pending synchronous operation waiting to unblock from the event.

• void HandleEvents(IOReadyEvents events);

  • This method is called for handling events that were returned by the method above

API Usage

These APIs are basically a minimum form of interface between the current SocketAsyncContext and SocketAsyncEngine. SocketAsyncContext would be the initial consumer: registration, unregistration, handling sync events inline, and handling events normally.

The APIs are not necessarily easy to use, they're meant to provide a minimal support for epoll/kqueue-based polling with handling events in thread pool threads. Here's some oversimplified pseudocode on how the APIs may be used, to read some data from a socket, process it, and write a response:

internal sealed class ClientSocketIOHandler : IIOReadyEventHandler
{
    SafeHandle _socketHandle;
    bool _isReading;
    byte[] _readBuffer;
    int _nextReadIndex;
    byte[] _writeBuffer;
    int _byteCountToWrite;
    int _nextWriteStartIndex;

    public ClientSocketIOHandler(SafeHandle socketHandle)
    {
        _socketHandle = socketHandle;
        _isReading = true;
        // Change the socket to non-blocking

        ThreadPool.RegisterForIOReadyNotifications(
            _socketHandle,
            IOReadyEvents.Read | IOReadyEvents.Write | IOReadyEvents.CloseOrError,
            this);
    }

    public IOReadyEvents HandleEventsInlineIfNecessary(IOReadyEvents events) => events;

    public unsafe void HandleEvents(IOReadyEvents events)
    {
        if ((events & IOReadyEvents.CloseOrError) != 0)
        {
            events |= IOReadyEvents.Read | IOReadyEvents.Write;
        }

        lock (this)
        {
            if ((events & IOReadyEvents.Read) != 0 && _isReading)
            {
                if (_readBuffer == null)
                {
                    _readBuffer = new byte[256];
                }

                var span = new Span<byte>(_readBuffer, _nextReadIndex, _readBuffer.Length - _nextReadIndex);
                int bytesReceived, errno;
                fixed (byte* b = &MemoryMarshal.GetReference(span))
                {
                    errno = Interop.Recv(_socketHandle, b, span.Length, 0, out bytesReceived);
                }

                if (bytesReceived <= 0) // error
                {
                    _isReading = false;
                    ThreadPool.UnregisterForIOReadyNotifications(_socketHandle);
                    return;
                }

                _nextReadIndex += bytesReceived;

                if (/* enough data read */) // done reading
                {
                    _isReading = false;
                    // Process the data, fill _writeBuffer
                    _readBuffer = null;
                    events |= IOReadyEvents.Write;
                }
            }

            if ((events & IOReadyEvents.Write) != 0 && _writeBuffer != null)
            {
                var span = new Span<byte>(_writeBuffer, _nextWriteStartIndex, _byteCountToWrite - _nextWriteStartIndex);
                int bytesSent, errno;
                fixed (byte* b = &MemoryMarshal.GetReference(span))
                {
                    errno = Interop.Send(_socketHandle, b, span.Length, 0, out bytesSent);
                }

                if (bytesSent <= 0) // error
                {
                    _writeBuffer = null;
                    ThreadPool.UnregisterForIOReadyNotifications(_socketHandle);
                    return;
                }

                _nextWriteStartIndex += bytesSent;
                if (_nextWriteStartIndex >= _byteCountToWrite) // done writing
                {
                    _writeBuffer = null;
                    ThreadPool.UnregisterForIOReadyNotifications(_socketHandle);
                    // Close the socket
                }
            }
        }
    }
}

Alternative Designs

• UnregisterForIOReadyNotifications - A separate method is proposed for unregistration rather than returning a disposable from the Register method. The expectation is that the event handler object would manage the lifetime of registration along with its own lifetime. A separate method for unregistration would avoid some unnecessary allocation for each IO handle that is registered.
• IOReadyEvents.CloseOrError - Alternatively, we could include all of the various bits (currently named ReadClose, Close, and Error in the implementation), translate to those exactly as how the system provides, and let the user disambiguate
• InternalsVisibleTo or reflection instead of making the APIs public - As far as I could tell, InternalsVisibleTo doesn't seem to be used anymore except for tests. Reflection is possible, though perhaps inconvenient.
• When thinking about adding APIs for io_uring it may make sense to put those APIs under System.IO (though with implementation in CoreLib), since it may need an API for each op code. Not sure if there would be a better place for these epoll/kqueue-based APIs than directly on the thread pool.

Risks

• Registering without unregistering would add to bookkeeping memory. Typically when all file descriptors representing a particular stream are closed, it would not receive events anymore, but there would be some bookkeeping cost from not unregistering. I feel it's a low risk, as it's a fairly low-level API.

Tagging subscribers to this area: @mangod9
See info in area-owners.md if you want to be subscribed.

Issue Details

---

Background and motivation

Currently we don't have a common way of handling IO readiness and completions across operating systems. On Windows the IO completion thread pool is not very efficient. The mechanism used on Linux in System.Net.Sockets is more efficient, but comes with other issues. As part of adding IO readiness/completion handling to the portable thread pool (transitioning away from the native thread pool), we would also like to consolidate the mechanisms used for IO readiness/completion handling across operating systems. Hopefully this would ease maintenance, make it easier for other components to also use the functionality, and enable further experimentation for improving things across the various async IO mechanisms, such as potentially with better thread management and better ordering of work.

The APIs being proposed here are specific to epoll/kqueue, analogous to the existing APIs for overlapped IO on Windows. Ideally, we would move to io_uring on Linux, but it's not available in all distros yet and we would continue to need a kqueue-based solution. When we decide to add support for io_uring, it would likely require a separate set of APIs.

More discussion here: #47631

API Proposal

namespace System.Threading
{
    public static class ThreadPool
    {
        [UnsupportedOSPlatform("windows")]
        [UnsupportedOSPlatform("browser")]
        public static void RegisterForIOReadyNotifications(
            SafeHandle handle,
            IOReadyEvents events,
            IIOReadyEventHandler eventHandler) { }

        [UnsupportedOSPlatform("windows")]
        [UnsupportedOSPlatform("browser")]
        public static void UnregisterForIOReadyNotifications(SafeHandle handle) { }
    }


    [Flags]
    public enum IOReadyEvents
    {
        None = 0x0,
        Read = 0x1,
        Write = 0x2,
        CloseOrError = 0x4
    }

    public interface IIOReadyEventHandler
    {
        IOReadyEvents HandleEventsInlineIfNecessary(IOReadyEvents events);
        void HandleEvents(IOReadyEvents events);
    }
}

RegisterForIOReadyNotifications

• Registers an IO handle to receive notifications of IO-ready events. The event handler is notified of IO-ready events for the IO handle for inline processing, or asynchronous processing on thread pool worker threads.
• This API is specific to Unix-like platforms and uses epoll or kqueue. The underlying implementation may not support some types of IO handles.
• Arguments
  • handle - The IO handle to register for IO-ready event notifications. The safe handle wraps a file descriptor that typically represents a non-blocking pipe or socket. An IO handle is keyed by the handle value and may be registered only once.
  • events - A mask of events for which to register for notifications
  • eventHandler - An object that would handle IO-ready events for the IO handle
• Exceptions
  • ArgumentNullException - One of the arguments is null
  • ArgumentException - handle is closed or invalid
  • ArgumentOutOfRangeException - The value of events is invalid
  • InvalidOperationException - handle was already registered
  • IOException - An error occurred when attempting to register the IO handle for notifications. The HResult property would contain an error code for further investigation

UnregisterForIOReadyNotifications

• Unregisters an IO handle to no longer receive notifications of new IO-ready events. Any IO-ready events that were already received prior to unregisteration are still delivered to the previously registered event handler.
• Arguments
  • handle - The IO handle to unregister for new IO-ready event notifications
• Exceptions
  • ArgumentNullException - One of the arguments is null
  • InvalidOperationException - handle was not registered

IOReadyEvents

• Read / Write - The stream is ready for a read or write respectively
• CloseOrError
  • There appear to be some inconsistencies in how epoll reports the RDHUP, HUP, and ERR events on Linux, see Add test to list OS event details tokio-rs/mio#1091 (comment). To simplify things, for epoll when any of those are set, just include this bit and have the user determine what happened based on the error code from the following IO operation.
  • For kqueue, similarly this bit would be set when EV_EOF or EV_ERROR are reported
  • System.Net.Sockets could continue to do what it's doing now and treat CloseOrError as Read | Write

IIOReadyEventHandler

• IOReadyEvents HandleEventsInlineIfNecessary(IOReadyEvents events);

  • Currently on Unix, when a synchronous operation is performed on a non-blocking socket, after the operation completes the blocking thread is released immediately upon receiving the event instead of queuing event processing to the thread pool, since otherwise the blocking may have occurred on a thread pool thread and lead to thread starvation
  • There is currently a config option to force processing events on the polling threads inline (along with configuring the number of polling threads), which we are currently planning to keep intact, at least for now
  • This method is called immediately upon receiving an event for inline processing
  • events specifies which IO-ready events were triggered
  • The returned events, if not None, will be queued for processing and will be sent later through HandleEvents. So, a user may choose to process all or some of the events inline depending on whether there is a pending synchronous operation waiting to unblock from the event.

• void HandleEvents(IOReadyEvents events);

  • This method is called for handling events that were returned by the method above

API Usage

These APIs are basically a minimum form of interface between the current SocketAsyncContext and SocketAsyncEngine. SocketAsyncContext would be the initial consumer: registration, unregistration, handling sync events inline, and handling events normally.

The APIs are not necessarily easy to use, they're meant to provide a minimal support for epoll/kqueue-based polling with handling events in thread pool threads. Here's some oversimplified pseudocode on how the APIs may be used, to read some data from a socket, process it, and write a response:

internal sealed class ClientSocketIOHandler : IIOReadyEventHandler
{
    SafeHandle _socketHandle;
    bool _isReading;
    byte[] _readBuffer;
    int _nextReadIndex;
    byte[] _writeBuffer;
    int _byteCountToWrite;
    int _nextWriteStartIndex;

    public ClientSocketIOHandler(SafeHandle socketHandle)
    {
        _socketHandle = socketHandle;
        _isReading = true;
        // Change the socket to non-blocking

        ThreadPool.RegisterForIOReadyNotifications(
            _socketHandle,
            IOReadyEvents.Read | IOReadyEvents.Write | IOReadyEvents.CloseOrError,
            this);
    }

    public IOReadyEvents HandleEventsInlineIfNecessary(IOReadyEvents events) => events;

    public unsafe void HandleEvents(IOReadyEvents events)
    {
        if ((events & IOReadyEvents.CloseOrError) != 0)
        {
            events |= IOReadyEvents.Read | IOReadyEvents.Write;
        }

        lock (this)
        {
            if ((events & IOReadyEvents.Read) != 0 && _isReading)
            {
                if (_readBuffer == null)
                {
                    _readBuffer = new byte[256];
                }

                var span = new Span<byte>(_readBuffer, _nextReadIndex, _readBuffer.Length - _nextReadIndex);
                int bytesReceived, errno;
                fixed (byte* b = &MemoryMarshal.GetReference(span))
                {
                    errno = Interop.Recv(_socketHandle, b, span.Length, 0, out bytesReceived);
                }

                if (bytesReceived <= 0) // error
                {
                    _isReading = false;
                    ThreadPool.UnregisterForIOReadyNotifications(_socketHandle);
                    return;
                }

                _nextReadIndex += bytesReceived;

                if (/* enough data read */) // done reading
                {
                    _isReading = false;
                    // Process the data, fill _writeBuffer
                    _readBuffer = null;
                    events |= IOReadyEvents.Write;
                }
            }

            if ((events & IOReadyEvents.Write) != 0 && _writeBuffer != null)
            {
                var span = new Span<byte>(_writeBuffer, _nextWriteStartIndex, _byteCountToWrite - _nextWriteStartIndex);
                int bytesSent, errno;
                fixed (byte* b = &MemoryMarshal.GetReference(span))
                {
                    errno = Interop.Send(_socketHandle, b, span.Length, 0, out bytesSent);
                }

                if (bytesSent <= 0) // error
                {
                    _writeBuffer = null;
                    ThreadPool.UnregisterForIOReadyNotifications(_socketHandle);
                    return;
                }

                _nextWriteStartIndex += bytesSent;
                if (_nextWriteStartIndex >= _byteCountToWrite) // done writing
                {
                    _writeBuffer = null;
                    ThreadPool.UnregisterForIOReadyNotifications(_socketHandle);
                    // Close the socket
                }
            }
        }
    }
}

Alternative Designs

• UnregisterForIOReadyNotifications - A separate method is proposed for unregistration rather than returning a disposable from the Register method. The expectation is that the event handler object would manage the lifetime of registration along with its own lifetime. A separate method for unregistration would avoid some unnecessary allocation for each IO handle that is registered.
• IOReadyEvents.CloseOrError - Alternatively, we could include all of the various bits (currently named ReadClose, Close, and Error in the implementation), translate to those exactly as how the system provides, and let the user disambiguate
• InternalsVisibleTo or reflection instead of making the APIs public - As far as I could tell, InternalsVisibleTo doesn't seem to be used anymore except for tests. Reflection is possible, though perhaps inconvenient.
• When thinking about adding APIs for io_uring it may make sense to put those APIs under System.IO (though with implementation in CoreLib), since it may need an API for each op code. Not sure if there would be a better place for these epoll/kqueue-based APIs than directly on the thread pool.

Risks

• Registering without unregistering would add to bookkeeping memory. Typically when all file descriptors representing a particular stream are closed, it would not receive events anymore, but there would be some bookkeeping cost from not unregistering. I feel it's a low risk, as it's a fairly low-level API.

Author:	kouvel
Assignees:	kouvel
Labels:	api-suggestion, area-System.Threading, untriaged
Milestone:	7.0.0

[kouvel]

CC @AntonLapounov @geoffkizer @karelz @mangod9 @stephentoub

[geoffkizer]

Looks reasonable to me. A couple related thoughts:

In the current SocketAsyncEngine code, the association between fd and event handler is managed using a ConcurrentDictionary. The dictionary lookup on each readiness notification adds a small amount of overhead. In theory, this shouldn't be necessary, as we should be able to associate user data (via a GCHandle) with the epoll registration when we call epoll_ctl. I believe we didn't do this because we ran in to issues with figuring out how to handle unregistration properly here. Would this proposal address that and remove the need for a dictionary lookup here?

See #43992

There is some weirdness in the current SocketAsyncEngine code with object references and finalization. Looking at your code example, I wonder if a similar issue exists... in particular, if the thread pool holds a reference to the registered IIOReadyEventHandler, then ClientSocketIOHandler will be kept alive and will never finalize. I think the solution here is just to make the reference to the IIOReadyEventHandler a weak ref. Would that work in your proposal?

See #43383

Also worth considering, some discussion about how we handle EPOLLRDHUP here: #43966

(edit to add links to relevant issues)

[tmds]

I don't think there are many users for this API, and that when you build something on top, you quickly run into the complexities (async state tracking) we deal with in SocketAsyncEngine.

Does this need to be a public API that is available for end-users?

I think by starting from the SocketAsyncEngine usage of epoll, you may also be limiting what this API allows you to do. For example, you can't do separate registrations to register for readable/writable.

[kouvel]

RE using a GC handle, I wasn't planning to include a fix for that but perhaps it can be done with a couple of things:

• When unregistering, issue an epoll_ctl to remove the fd so that it does not get further notifications
• Don't dispose the GC handle upon unregistering, rather track it as a dead handle and have the corresponding epoll thread dispose them before the next wait to avoid race conditions

There is some weirdness in the current SocketAsyncEngine code with object references and finalization.

I'm not too familiar with the specific issues mentioned in #43383, but my understanding is that the issue with rogue references in SocketAsyncEngine was due to the infinite-loop polling method where the JIT was artificially extending the lifetimes of some locals. I think we can avoid the rogue references to the event handler in a similar way for now. It may be a non-issue if we don't use dedicated threads for polling, though that looks to be difficult to achieve with epoll due to substantial overhead in polling.

Also worth considering, some discussion about how we handle EPOLLRDHUP here: #43966

That's interesting. I suppose if we get a CloseOrError we can skip the optimization?

A strange thing for a pipe was this based on the link in the CloseOrError description:

Unix pipe sender, receiver closed:
	EPOLLOUT|EPOLLERR

Where there's no HUP even though it is a HUP situation and not an ERR situation, which is partly why I suggested merging the close/error cases to simplify things. Also, ERR doesn't help much, as it doesn't provide an error code, so I'm not sure it's worth separating the Close/Error conditions.

I don't think there are many users for this API, and that when you build something on top, you quickly run into the complexities (async state tracking) we deal with in SocketAsyncEngine.

Does this need to be a public API that is available for end-users?

Agreed. I don't think it needs to be public to start with, however, I didn't find a case where we have non-public APIs that can be used between libraries. Previously that's what InternalsVisibleTo was used for, but I don't see it being used anymore for that purpose. If there is a sanctioned way to do that, I would be fine with making these APIs non-public.

I think by starting from the SocketAsyncEngine usage of epoll, you may also be limiting what this API allows you to do. For example, you can't do separate registrations to register for readable/writable.

From what I gathered, epoll_ctl with EPOLL_CTL_ADD twice on the same fd would result in EEXIST. A restriction in the proposal is that one cannot modify a registration. If that were to become necessary, it could be added. Can you elaborate on what you mean by separate registrations and why it would be done?

[tmds]

I didn't find a case where we have non-public APIs that can be used between libraries. Previously that's what InternalsVisibleTo was used for, but I don't see it being used anymore for that purpose. If there is a sanctioned way to do that, I would be fine with making these APIs non-public.

I'm curious to learn this too.

If we use this API to get rid of the artificial dependency on Socket (#47631), we'll be duplicating SocketAsyncContext to something similar in System.IO.Pipes. It may be nice if we could share more code between these epoll-users instead of duplicating it.

If that were to become necessary, it could be added. Can you elaborate on what you mean by separate registrations and why it would be done?

It would allow to keep the code paths for reading and writing separate.
Socket uses epoll in a specific way, and that is shaping the behavior of this API.

It is less important to consider potential end-user use-cases if the API is internal.

[danmoseley]

Agreed. I don't think it needs to be public to start with, however, I didn't find a case where we have non-public APIs that can be used between libraries. Previously that's what InternalsVisibleTo was used for, but I don't see it being used anymore for that purpose. If there is a sanctioned way to do that, I would be fine with making these APIs non-public.

It is harder to reason about dependencies if there are private relationships. It is possible it might break assumptions in tooling as well (such as trimming? idk).

So far we have managed by either using shared code or by finding some public surface we're comfortable with or by forwarding types. are any of those possible here.

[jkotas]

I didn't find a case where we have non-public APIs that can be used between libraries

A few options:

• API is public in CoreLib, but not exposed in reference assemblies (e.g. Debug.SetProvider use in System.Diagnostics.TraceSource is on this plan)
• API is marked [RequiresPreviewFeaturesAttribute] if we think that the API will be public eventually, but we are not sure about the exact shape.

[stephentoub]

Thanks for the issue, @kouvel.

Have we proved out the perf benefits that would come from integrating at this level? The two main benefits that are likely to drive this are improved perf and improved ability to utilize this functionality from lower in the stack to achieve asynchrony, e.g. from FileStream, but most of the places we'd want to use this either won't work with this mechanism anyway (e.g. epoll doesn't work well with regular disk file descriptors) or have already been replat on top of Socket as a hack to get its functionality. So I expect the primary benefit here would really be about performance, and we should prototype / validate the wins prior to pushing for any new surface area here (if it's an implementation detail that's not part of the surface area, we have more flexibility).

I've also hoped that we'll be able to utilize io_uring in Sockets, ideally starting in .NET 7. While it'd only be usable on newer kernels, such kernels should be more commonplace by the time we ship .NET 7, and I'm not sure adding this epoll/kqueues mechanism in support of "legacy" will then be worthwhile. Maybe.