proposal: net, os: API to interrupt Read/Write() for polled objects

[networkimprov]

Problem

Although we can interrupt/pause the I/O poller via c.SetDeadline(time.Now()), that's not reliable if Set*Deadline() is called for other reasons. Interrupting a poller is easy to do via native OS APIs.

Consider a message receiver, started via go runRecv(c). It's easy to follow and needs 1 goroutine.

func runRecv(iConn net.Conn) {
   ...
   for {
      iConn.SetReadDeadline(...)
      aLen, err := iConn.Read(...)
      if err != nil {
         // handle errors
         return
      }
      // handle input; may entail Read()
   }
}

Support out-of-band input. Now it's hard to follow and needs 2 goroutines, 2 channels, and a select.
Note: this use case has been critiqued at length below; please don't rehash those points.

func runRecv(iConn net.Conn, iAlt chan T) {
   aReadFlag := make(chan bool)
   var aLen int
   var err error

   go func() {
      for <-aReadFlag { // wait for handler
         iConn.SetReadDeadline(...)
         aLen, err = iConn.Read(...)
         aReadFlag <- true
      }
   }()
   defer func() { aReadFlag <- false }()

   for {
      aReadFlag <- true // signal to reader
   WaitForMsg:
      select {
      case aMsg := <-iAlt:
         // handle out-of-band input
         goto WaitForMsg
      case <-aReadFlag:
      }
      if err != nil {
         // handle errors
         return
      }
      // handle network input; may entail Read()
   }
}

Proposal

Let us interrupt Read/Write() for polled objects. This is easy to follow and needs 1 goroutine.

func runRecv(iConn net.Conn) {
   ...
   for {
      iConn.SetReadDeadline(...)
      aLen, err := iConn.Read(...)
      if err != nil {
         if err == io.Stopped { // iConn.Stop() was called
            // handle out-of-band input
            continue
         }
         // handle errors
         return
      }
      iConn.SetReadStop(false)
      // handle network input; may entail Read()
      iConn.SetReadStop(true)
   }
}

API

As methods of polled types
func (o *T) Stop() error causes the pending or next o.Read/Write() to return io.Stopped.
func (o *T) SetStop(s bool) error false prevents or postpones io.Stopped result.
func (o *T) SetReadStop(s bool) error false prevents or postpones io.Stopped result from Read().
func (o *T) SetWriteStop(s bool) error false prevents or postpones io.Stopped result from Write().

Alternative functions in package io
func Stop(o interface{}) error
func SetStop(o interface{}, s bool) error
func SetReadStop(o interface{}, s bool) error
func SetWriteStop(o interface{}, s bool) error

Alternative that accommodates Context.Done(); channel send or close works like Stop() above
func (o *T) SetStopper(c <-chan struct{}) error nil prevents io.Stopped result.
func (o *T) SetReadStopper(c <-chan struct{}) error
func (o *T) SetWriteStopper(c <-chan struct{}) error

@ianlancetaylor suggested a modeless API
func (o *T) StoppableRead(b []byte) (int, error) can return io.Stopped.
func (o *T) StoppableWrite(b []byte) (int, error) can return io.Stopped.

Multiple Stop() requests outside of polling or after Set*Stop(false) yield at most a single io.Stopped error.

If Stop() is called after Set*Stop(false), a Set*Stop(true) causes the pending or next Read/Write() to return io.Stopped. Alternatively Set*Stop() could take an argument indicating never, postpone, or asap, where never causes subsequent Stop() requests to be ignored instead of postponed.

The API supports each polled type, e.g. net.TCPConn, net.UnixConn, os.File. The calls return an error if the object is not open/connected.

cc @aclements @rsc @Sajmani

[networkimprov]

Related #20280 - using Context to interrupt Read/Write()

[robaho]

i think this would lead to poor code over using channels. Now you have a reader that must handle the OOB input as well. Also, it complicates partial messages as this might need to be accounted for in handling the OOB message. Seems like it will lead to a lot of concurrency issues.

Contrast with the simple solution that the io reader places completed messages on a channel. OOB messages can be placed on the same channel. A command processor reads the channel. No api changes and a cleaner responsibility structure.

[networkimprov]

The issue is pausing the I/O poller (a general purpose feature), not out-of-band input for network streams. I've clarified that in the problem statement.

What you describe is just slightly more efficient than what I wrote, and doesn't help if receiving a complete messages requires fully processing it.

[bcmills]

Support out-of-band input. Now it's hard to follow and needs 2 goroutines, 2 channels, and a select.

As far as I can tell, a simpler implementation of that logic requires only one extra goroutine and one channel, although I find it clearer with two channels (https://play.golang.org/p/UMk09ufmAU9). It is straightforward to add a third (1-buffered) channel to eliminate the buffer reallocation in the steady state, if you are so inclined.

Either way, it seems much clearer to put the Read results themselves on the channel instead of using it as at out-of-band control channel. (This is closely related to the recurring “share by communicating” theme of my GC '18 talk.)

Note that with that approach, the Read goroutine may complete an already-pending Read call while the consumer is blocked on an interrupt, but won't begin another Read — and will therefore reduce pressure on the poller — until the consumer has received the result.

[robaho]

I think @bcmills code is very simple to understand, and addresses the 'pausing' requirements, and partial message issues (that can be VERY difficult to handle).

[networkimprov]

Again, the point is pausing the I/O poller, not implementing out-of-band input per se.

For reference, here's the core of @bcmills concept, which needs 2 goroutines, 3 channels, and 2 selects.

As before, it doesn't help if receiving a complete message requires fully processing it (e.g. the message body is streamed to disk), which I flagged via // handle network input; may entail Read().

func logAll(ctx context.Context, nc net.Conn, ic <-chan interruption) error {
	defer nc.Close()

	var (
		readPacket = make(chan []byte)
		readErr    = make(chan error, 1)
	)
	go func() (err error) {
		defer func() { readErr <- err }()
		for {
			buf := make([]byte, maxPacketSize)
			n, err := nc.Read(buf)
			if err != nil {
				return err
			}
			select {
			case <-ctx.Done():
				return ctx.Err()
			case readPacket <- buf[:n]:
			}
		}
	}()

	for {
		select {
		case err := <-readErr:
			if err != io.EOF {
				return err
			}
			return nil
		case packet := <-readPacket:
			log.Printf("Read: %s", packet)
		case interrupt := <-ic:
			log.Printf("Interrupt: %s", interrupt)
		}
	}
}

[bcmills]

Why is it important to pause the poller per se?

[ianlancetaylor]

I also like @bcmills 's solution. Why confuse matters with the notion of out-of-band data and stopping the poller? What do you gain from that? My best guess for what you mean by "out-of-band data" is that you have two different input streams: one from the network connection and one from something else. In Go the natural way to handle two different input streams is to put them on a channel and use select. I don't yet see the compelling argument for doing anything else.

[robaho]

But even then, if pausing the poller is the “goal”, when reading a message and streaming to disk? this would be done in a loop anyway (for arbitrary length messages) - a simple Lock() would allow the poller to be paused (the external interrupter takes the lock, pausing the poller). I think the solution outlined by @bcmills is the “go” way (and the “proper” way as well).

…

On Jan 6, 2020, at 4:33 PM, Bryan C. Mills ***@***.***> wrote: Why is it important to pause the poller per se? — You are receiving this because you commented. Reply to this email directly, view it on GitHub <#36402?email_source=notifications&email_token=ABF2U4NHF7LJ2EFPUGV5QVLQ4OWSVA5CNFSM4KDANSZKYY3PNVWWK3TUL52HS4DFVREXG43VMVBW63LNMVXHJKTDN5WW2ZLOORPWSZGOEIHAR7Q#issuecomment-571345150>, or unsubscribe <https://github.com/notifications/unsubscribe-auth/ABF2U4LRYJE5EIOARE3WCRTQ4OWSVANCNFSM4KDANSZA>.

[networkimprov]

With an OS-native API, it's easy to code anything which requires that certain polled objects aren't in-progress. Virtually inserting a message into a TCP stream is but one example.

Sure, we can approximate this today, but the result is harder to read and less efficient. My proposed example (pasted below) is notably clearer and more efficient than @bcmills suggestion, and trivially supports very long messages.

EDIT: We can already stop pause the poller with a timer via Set*Deadline(), so why not with a call?

func runRecv(iConn net.Conn) {
   ...
   for {
      iConn.SetReadDeadline(...)
      aLen, err := iConn.Read(...)
      if err != nil {
         if err == io.Stopped { // iConn.Stop() was called
            // handle out-of-band input
            continue
         }
         // handle errors
         return
      }
      iConn.SetReadStop(false)
      // handle network input; may entail Read()
      iConn.SetReadStop(true)
   }
}

[ianlancetaylor]

I either don't understand your example or I don't understand the proposal. It seems to me that the call to iConn.SetReadStop(true) at the end of the loop means that the call to iConn.Read at the start of the loop will always return io.Stopped. Can you explain how this is supposed to work?

[robaho]

The SetReadStop(true), means that future Stop() calls will work on reads, etc. If it is false, then Stop() is ignored during read.

@networkimprov I think a problem with your proposal is that you are taking a polled approach. Go uses 'select/poll' behind the scenes to be efficient, but a different implementation might not do it that way. What Go exposes is a simpler serial interface - protecting the developer from having to deal with the intricacies of select/poll - in some ways you are adding that back in. If you could 'select' on a Conn, that would be a different story - and a completely differently interface.

Another problem with your code is that it essentially limits you to a single "interrupter/OOB supplier", or you need some additional advanced state management (bringing the overall complexity up). The @bcmills version does not have this limitation.

As a sidenote, Ian mentioned that the SetDeadline() takes effect for existing Reads - so you could set a flag and set a very small read timeout and accomplish the goal.

[networkimprov]

@ianlancetaylor my example lacks an iConn.Stop() (see comment); it occurs in a different goroutine in this case. Only Stop() causes err == io.Stopped.

The iConn.SetReadStop(true) causes io.Stopped on subsequent Read() only if Stop() occurred since the last Read().

@robaho you can use any means (incl select{...}) to reach an alternate event supplier on io.Stopped. I addressed the c.SetDeadline(long_past) concept in the golang-dev thread:

Unfortunately you have to clear the long-past deadline (or set in future) before each Read/Write() so if the SetDeadline(long_past) call arrives after the I/O op returned, you can't see it. Also the resulting error is the same for long-past deadlines, ordinary deadlines, and keepalive failures.

This is clearly a variation on Set*Deadline(), not a polling API.

[robaho]

@networkimprov upon more detailed reading, this is purely a code structure problem not an API/functionality problem. You can't read a partial message - the data is still in the stream - so it must be read eventually - your problem is that it is delaying the handling of 'external OOB' - but that is only because you have a single routine attempting to process 2 sources - there is no reason to do this in Go, as the routines are cheap - @bcmills code is structured far cleaner.

Like I said, your code because much more complex if there is more than one provider of OOB messages - with @bcmills it is a trivial and straightforward change.

I don't think your code is cleaner. It clearly breaks the single responsibility principle - you have a function performing IO and processing messages from other actors (nested within the IO processing).

As an aside, I've only ever seen ReadDeadline() used to terminate a connection when a heartbeat doesn't arrive (dead TCP stream detection), using it for anything else is troublesome and a poor design choice IMO.

[networkimprov]

This is intentional: "it is delaying the handling of 'external OOB'". Thanks anyway.

[rogpeppe]

@networkimprov

We can already stop the poller with a timer via Set*Deadline(), so why not with a call?

You're not proposing a way to stop the poller, but to pause it, which is a very different thing.

Although we can interrupt/pause the I/O poller with a timer via Set*Deadline(), there's no way to do that with a call -- something which is easy to do via native OS APIs.

It might be useful to show the actual C code that would be easy to do with a native OS API, and describe why, for your particular use case, the existing solutions aren't sufficient.

[ianlancetaylor]

Sorry for misunderstanding how this API works.

I don't yet see the point of this. Go supports having a single goroutine handle two different input sources by using channels and select statements. What do we gain by providing another mechanism? The resulting code looks harder to understand and more prone to race conditions. And the overall approach is hard to explain. What is the benefit? We just get to save a goroutine? But goroutines are relatively cheap.

[rogpeppe]

@ianlancetaylor The only thing I can think of is that there's no way to stop an in-progress read call, issue some other system call on the file descriptor, then continue reading. I'm not familiar with the OOB API, but perhaps the idea is that you get a signal indicating that OOB data is present, stop any existing read call, then issue a different kind of syscall to get the OOB data, and then a normal Read call again. Currently I don't see that there's a way to do that (@bcmills' solution would still involve a race on the underlying fd). This is why I think it would be useful to know the actual problem that the original poster is trying to solve.

[tv42]

If this is about MSG_OOB as per recvmsg(2) and friends, neither of the designs here is appropriate and Go already has the right low-level building blocks:
https://godoc.org/golang.org/x/sys/unix#Recvmsg
https://godoc.org/golang.org/x/sys/unix#Sendmsg

[robaho]

The OPs use of OOB is not related to TCP "out of band" data, nor TCP/UDP "ancillary data (e.g. timestamps)".

By OOB, he is referring to a interruption of serial processing to handle an async event.

[rogpeppe]

Are there any use cases other than OOB data that this API might be useful for?

[ianlancetaylor]

One consideration here is that setting a mode is racy in a concurrent program, and in general we should avoid modes whenever possible. So rather than SetStop we should have StoppableRead. Then I guess Stop would mean that the current or next StoppableRead would return io.Stopped. But if we express the problem in that way, then Stop seems isomorphic to SetDeadline(time.Now()).

[networkimprov]

@rogpeppe thanks for thinking about ways this could be used outside my example. Re "not ... a way to stop the poller, but to pause it," this has the same effect as Set*Deadline(), but with the error value io.Stopped.

Re what we can do in C, I haven't used the epoll API for a while, but off the top of my head, you can add a a unix-socket to its fd list and interrupt epoll_wait() by pinging it. Then you can do anything before resuming epoll_wait().

The code in the issue which I dubbed "hard to follow and needs 2 goroutines, 2 channels, and a select" is what I'm actually doing. It works, but it made me wish for a variant of Set*Deadline() which can be triggered by a call.

@ianlancetaylor iConn.StoppableRead(...) is an interesting idea, thanks. Note that Stop() affects either the pending or next StoppableRead().

iConn.SetDeadline(time.Now()) doesn't work like iConn.Stop() because, as mentioned above, "you have to clear the long-past deadline (or set in future) before each Read/Write() so if the SetDeadline(long_past) call arrives after the I/O op returned, you can't see [a timeout error on next I/O op]. Also the resulting error is the same for long-past deadlines, ordinary deadlines, and keepalive failures."

[robaho]

This is sounding an awful lot like "interrupts" in Java. I suggest if this proposal was to move forward, it would adopt similar syntax and semantics. e.g. Interrupt(), IsInterrupted() etc. This would work on any blocked Go routine.I will say that a big part of the reason Java needs this is that the threads are heavy, so you need to conserve their usage. Similar needs led to async IO facilities. Go took a different (and better approach IMO), and this proposal seems at odds with the Go way of spinning up another routine. I suggest we don't muddy the waters with another way to accomplish the same thing.

[robaho]

On review, the "interrupts" semantics of Java cannot be applied to Go, as there is no reference available to the routine you would want to interrupt, so Interrupt() would need to be on the Conn, and probably any other blocking resource if this was to be "complete".

Seems way too many changes in this case, when the current method works fine IMO.

(In the OPs case, what if the resource was a network file descriptor (or other "slow" device, or a call that blocked until a "device status change" - which might be indefinite), you would need the same Stop, etc. on all of these resource handles.

[rogpeppe]

The code in the issue which I dubbed "hard to follow and needs 2 goroutines, 2 channels, and a select" is what I'm actually doing. It works, but it made me wish for a variant of Set*Deadline() which can be triggered by a call.

If that works, I don't understand why you need to interrupt the read at all. Why not just issue the syscall concurrently with the read and not bother with any of the mechanism? AFAICS you've always got the risk of that happening anyway.

[networkimprov]

I'm not making a syscall, so not sure what you mean... I need to either prevent the goroutine that calls Read() from processing received data when the poller wakes it if there's an alternate-source message waiting (what I do now), or force it awake to handle the alternate source by a mechanism like iConn.Stop().

Since the latter might be generally useful, and also satisfy #20280, I've proposed it here.

[networkimprov]

Re "already has [an additional goroutine,] the one that would be calling Stop()," that's required by the other example too, for alt_channel <- message. Don't casually accuse ppl of intellectual dishonesty.

@rogpeppe you've repeated @bcmills concept, summarized in #36402 (comment). I've repeatedly stated that it doesn't work for long messages which must be fully processed (e.g. streamed to disk) to be "complete".

For the nth time, this isn't a proposal re insertion of messages into streams from alternate sources. Could we please discuss the applications and problems of an API to interrupt a polled object?

[bcmills]

Could we please discuss the applications and problems of an API to interrupt a polled object?

@networkimprov, it seems that none of the rest of us are seeing applications that require this behavior. Could you provide some examples of concrete, existing programs where this would be useful, where the alternate approach is not satisfactory?

[networkimprov]

Another use case is #20280 which seeks to interrupt polled objects on Context cancellation. I ref'd it in the first comment above :-)

[ianlancetaylor]

@networkimprov #20280 also doesn't give any examples of existing programs. It refers to #18507, and #18507 has been fixed since #20280 was proposed. So unfortunately that reference isn't too helpful.

This API you are proposing is inherently racy, since it requires coordination between the goroutine calling Read and the goroutine calling Stop; consider what happens if the Read completes just as a goroutine calls Stop. It can only be used correctly in cases where that doesn't matter, which means that it is relatively subtle. And it duplicates the existing SetDeadline mechanism which has the same problems.

Given these complexities I think we need a clear explanation of significant benefit that outweighs the cost, and I don't think we've seen one.

[networkimprov]

I don't see how coordination between goroutines using the API is necessary or useful. Stop() causes io.Stopped from the pending or next Read/Write(); it doesn't promise the earliest conceivable one. It does promise that io.Stopped will occur when calling Read/Write() in a loop, which is by far the common case. That also applies to SetDeadline().

When not calling Read/Write() in a loop, you don't rely on io.Stopped to do something; you simply do it after Read/Write() returns.

There is recent activity on #20280 by @navytux re https://github.com/navytux/go123,
and this experience report by @zombiezen covers an existing program:
https://medium.com/@zombiezen/canceling-i-o-in-go-capn-proto-5ae8c09c5b29

EDIT: Also the example I gave for "Support out-of-band input" is from a released application.

Lacking Stop(), SetReadDeadline() allows frequent polling of an "interrupt" flag, but that interferes with using deadlines to monitor connection health or protocol correctness, and creates overhead, i.e.
a) wake the polling thread
b) schedule the blocked goroutine
c) block it again
d) sleep the polling thread

[rogpeppe]

@networkimprov

I've repeatedly stated that it doesn't work for long messages which must be fully processed (e.g. streamed to disk) to be "complete".

Yes, you've repeatedly stated that, but you haven't explained any further and I don't understand why that's the case. Why can't the goroutine that's receiving the data from the network reader also stream that data to disk when appropriate?

[networkimprov]

It's moot, but streaming is in pkg.HandleMessage() which you wanted in the "muxing" thread. If you stream in the "reading" thread, it conflicts with muxing work.

[ianlancetaylor]

@networkimprov If the goroutines don't coordinate, then this API only works reliably if one goroutine calls Read and Write in a loop, as you say. That is a subtle point.

The experience report by @zombiezen does not seem to be the what you are talking about. That article is how safely closing down a connection. We've worked to eliminate those problems, by making it reliable to concurrently call Close and Read, and by adding support for SetDeadline on *os.File. Since we have done that work, this proposal would not help with that problem.

With regard to the xio package, I have no problem with using a context.Context to cancel a Read or Write. That also is not this proposal.

[networkimprov]

[reposted with major revisions]

Your quote from #20280 (comment) (after #18507 was fixed) is one inspiration for this proposal:

"I think we can ... focus on the idea of whether it is possible to arrange for a Read to be interrupted, with an error, when some Context is canceled. We can implement that for descriptors that end up in the network poller ... Can we devise a mechanism for associating a Context with a *os.File or net.Conn?"

I drafted a more general-purpose API, as not everyone uses Context. How could a polling interrupt scheme via Context result in a less "subtle" API?

Thinking aloud, we could define this on pollable types to let channel send or close cause interrupt.
func (o *T) SetStopper(c <-chan struct{}) error
The pending or next Read/Write() after channel send returns err == io.Stopped.

This enables iConn.SetStopper(ctx.Done()), eliminating the extra goroutine described by @zombiezen in the "What Doesn't Work" section of his report.

this API only works reliably if one goroutine calls Read and Write in a loop

To clarify, either a Read() loop, or Write() loop; not Read() and Write() in same loop. And it's entirely reliable, but like SetDeadline() or any I/O routine, it's non-deterministic.

[ianlancetaylor]

I think we should encourage people to use context.Context. I don't think we need a different API for cancelation. One is enough.

Also I suggested associating a context.Context with a file or network connection. I don't know if that is a good idea. But it's not the idea in this issue. I was thinking in terms of "if this context is canceled, we no longer care about this network connection." But this proposal is about something different.

[robaho]

Without beating a dead horse, every Go routine should have an implicit context (TLS again...), but in order to use that effectively, you need a reference to the Go routine (at least traditionally - you could use the Context reference as a substitute for the routine reference). That is

theContext := go somefunc()

Then you can do things like, theContext.setName(), etc.
But you probably want to set some of these properties at creation time to avoid race conditions, via

theContext := go "somename" somefunc()

etc.

Then it is a simple matter of changing the internal runtime code to access the routines context during scheduling/io to know if the operation has been cancelled. Clearly Context.Cancel() would need some internal runtime support to interrupt outstanding IO using that context. Context would also need to be thread safe.

You could probably also do something like this to share contexts:

go sharedcontext somefunc()

So a Cancel() on the Context could abort IO/cancel in any related Go routines.

A course grained Context would be easier to manage and still provides LOTS of value. I realize this is different than the OPs request but since the door was opened... I still can't find a valid use-case for the OPs request...

[networkimprov]

That's off-topic; pls file a new issue, thanks.

[robaho]

It was to clarify that @ianlancetaylor recommendation to use context.Context has some gaps IMO.

[networkimprov]

@ianlancetaylor it would be strange to add a feature to the runtime that allows polled objects to be interrupted non-terminally and not provide any API to it except context.With*().

But in any case, please add the feature, and in such a way that any reasonable API can trigger it. Then any project can add an app-specific API to it. (Many of us already patch the stdlib to make Go work for our apps.)

Should I file a new issue for a runtime feature?

[ianlancetaylor]

Should I file a new issue for a runtime feature?

We don't have any agreement on #20280. I don't think we'll get any agreement on a more general feature.

[networkimprov]

To clarify, I mean an internal polling-interrupt capability, with public API to be determined later.

This would allow projects to experiment with APIs, and likely lead to consensus re a public API.

[ianlancetaylor]

We already have an internal polling-interrupt capability: runtime.netpollunblock.

We aren't going to add another one without understanding what the public API will look like first. What you are suggesting sounds like the opposite of the way we do things.

[networkimprov]

Is this the mechanism by which a deadline timer unblocks its object on timeout?
//go:linkname poll_runtime_pollUnblock internal/poll.runtime_pollUnblock
func poll_runtime_pollUnblock(pd *pollDesc)

If so, how is the Read/Write() error set to indicate SetDeadline() timeout?

Could I make a patch with an exported function that sets a different error value, and then invokes the above call?

[ianlancetaylor]

No, runtime.pollUnblock is used when a descriptor is being closed, to wake up reads and writes blocking on that descriptor.

Deadline timers unblock via poll_runtime_pollSetDeadline aka internal/poll.runtime_pollSetDeadline which calls netpollunblock.

[rsc]

This proposal has been declined as infeasible.
— rsc for the proposal review group