fn: experimental async API #8234
Conversation
fn/async.go
Outdated
| } | ||
|
|
||
| func (a *async[A]) Cancel() { | ||
| close(a.quit) |
There was a problem hiding this comment.
This is the main thing I'm concerned about. If we tie the Async action to a larger context then the Cancel function will cancel the whole context. So it's possible that RunAsyncWithCancel is an unsafe API that we don't want to expose.
There was a problem hiding this comment.
With the current API, isn't async.quit always a newly created chan? How exactly would quit get tied to some other chan?
Edit: I see, RunAsyncWithCancel is exposed...
There was a problem hiding this comment.
Probably we want two chans -- one to cancel from outside, and one to cancel by calling Cancel.
There was a problem hiding this comment.
🤔 hmm interesting.
|
Concept ACK. I think this could remove a lot of boilerplate from the codebase. |
ProofOfKeags
force-pushed
the
feature/async
branch
from
5e32d2b
to
a17d1f7
Compare
ProofOfKeags
force-pushed
the
feature/async
branch
from
a17d1f7
to
65137c0
Compare
fn/async.go
Outdated
| // and make it a cancellable asynchronous call. This can be useful if the | ||
| // function in the argument can block indefinitely and the calling context needs | ||
| // a way to abandon waiting for it to complete. | ||
| func RunAsyncWithCancel[A any]( |
There was a problem hiding this comment.
I think it's worth adding a test for this
There was a problem hiding this comment.
I'll write the test after we come to consensus on the desired API.
There was a problem hiding this comment.
I happened to write a test for testing the suggested change in the other comment. maybe it's useful so just gonna paste here
func TestAsync(t *testing.T) {
// Let test func fail internally.
tf := newTestFunc()
async := RunAsync[bool](tf.Run)
tf.forceError()
_, err := async.Await()
require.ErrorContains(t, err, "testFunc error")
// Let the test func succeed.
tf = newTestFunc()
async = RunAsync[bool](tf.Run)
tf.complete()
_, err = async.Await()
require.NoError(t, err)
// Force the func to exit due to parent context being cancelled.
tf = newTestFunc()
ctx := context.Background()
ctxc, cancel := context.WithCancel(ctx)
async = RunAsyncWithCancel[bool](ctxc, tf.Run)
cancel()
_, err = async.Await()
require.ErrorContains(t, err, "context canceled")
// Force the func to exit due to an external Cancel call.
tf = newTestFunc()
async = RunAsync[bool](tf.Run)
async.Cancel()
_, err = async.Await()
require.ErrorContains(t, err, "context canceled")
}
func newTestFunc() *testFunc {
return &testFunc{
error: make(chan struct{}),
succeed: make(chan struct{}),
}
}
type testFunc struct {
error chan struct{}
succeed chan struct{}
}
func (f *testFunc) Run(ctx context.Context) (bool, error) {
select {
// Exit if parent ctx is cancelled.
case <-ctx.Done():
return false, ctx.Err()
case <-f.succeed:
return true, nil
case <-f.error:
return false, fmt.Errorf("testFunc error")
}
}
func (f *testFunc) complete() {
close(f.succeed)
}
func (f *testFunc) forceError() {
close(f.error)
}
fn/async.go
Outdated
| err: make(chan error), | ||
| } | ||
| go func() { | ||
| res, err := f() |
There was a problem hiding this comment.
should we allow the function to take in a context.Context so that we can cancel it if Cancel is called?
There was a problem hiding this comment.
Yeah I think this is a good consideration. I don't have much experience using context.Context. The trick would be that we would need to make the argument function accept a Option[Context], or maybe add a second API that allows us to thread a function that accepts a context in.
There was a problem hiding this comment.
just change the current argument to f func(ctx context.Context) (A, error)?
There was a problem hiding this comment.
I think context.Context can do a lot of the work for you here actually. Wdyt of this:
type Async[A any] interface {
Await() (A, error)
Cancel()
}
func RunAsyncWithCancel[A any](ctx context.Context,
f func(ctx context.Context) (A, error)) Async[A] {
ctxc, cancel := context.WithCancel(ctx)
a := async[A]{
ctx: ctxc,
res: make(chan A, 1),
err: make(chan error, 1),
cancel: cancel,
}
a.wg.Add(1)
go func() {
defer a.wg.Done()
res, err := f(ctxc)
if err != nil {
a.err <- err
return
}
a.res <- res
}()
return &a
}
func RunAsync[A any](f func(ctx context.Context) (A, error)) Async[A] {
return RunAsyncWithCancel(context.Background(), f)
}
func (a *async[A]) Await() (A, error) {
var aa A
select {
case r := <-a.res:
return r, nil
case e := <-a.err:
return aa, e
case <-a.ctx.Done():
return aa, a.ctx.Err()
}
}
func (a *async[A]) Cancel() {
a.cancel()
a.wg.Wait()
}
type async[A any] struct {
ctx context.Context
cancel func()
res chan A
err chan error
wg sync.WaitGroup
}There was a problem hiding this comment.
Seems reasonable to me.
ProofOfKeags
force-pushed
the
feature/async
branch
from
65137c0
to
c7b5ddd
Compare
|
Naively, the API looks okay to me. I agree with @Crypt-iQ though, we need to see how this is actually used at call sites to inform our API design. |
ProofOfKeags
force-pushed
the
feature/async
branch
from
c7b5ddd
to
b30fddf
Compare
|
Important Review skippedAuto reviews are limited to specific labels. Labels to auto review (1)
Please check the settings in the CodeRabbit UI or the You can disable this status message by setting the Thank you for using CodeRabbit. We offer it for free to the OSS community and would appreciate your support in helping us grow. If you find it useful, would you consider giving us a shout-out on your favorite social media? TipsChatThere are 3 ways to chat with CodeRabbit:
Note: Be mindful of the bot's finite context window. It's strongly recommended to break down tasks such as reading entire modules into smaller chunks. For a focused discussion, use review comments to chat about specific files and their changes, instead of using the PR comments. CodeRabbit Commands (invoked as PR comments)
Additionally, you can add CodeRabbit Configuration File (
|
| } | ||
|
|
||
| go func() { | ||
| res, err := f(localCtx) |
There was a problem hiding this comment.
so the provided function needs to cleanly listen to the Done() channel to actually be abortable ?
There was a problem hiding this comment.
If you do not listen on context.Done, then the goroutine will hang on whatever other channel signal it is waiting on to continue. The control flow of the Async itself will remain cancellable via the Cancel function producing an error for the thread that is hanging on Await, but a cancellation of the async action won't necessarily abort the inner thread unless f listens on context.Done
There was a problem hiding this comment.
That's the best we can do I guess, and the caller needs to be responsible as well.
| if err != nil { | ||
| select { | ||
| case a.err <- err: | ||
| case <-a.ctx.Done(): |
There was a problem hiding this comment.
Not an expert but feels designwise a bit odd to always check for the cancel channel and not summarize the cases.
There was a problem hiding this comment.
I'm not sure what you mean here. I made this choice since we don't want to spend effort on sending back over the error channel if we've already been cancelled. I suppose that since the channel is buffered and we only ever send on it once, that this shouldn't ever happen though.
There was a problem hiding this comment.
I meant the two select statement could be maybe summarized right. Or even go with the proposal of elle which did not use the Done() channel at all because we know they are buffered and will not hold up the process.
a.wg.Add(1)
go func() {
defer a.wg.Done()
res, err := f(ctxc)
if err != nil {
a.err <- err
return
}
a.res <- res
}()
| // function in the argument can block indefinitely and the calling context needs | ||
| // a way to abandon waiting for it to complete. | ||
| func RunAsyncWithCancel[A any](ctx context.Context, | ||
| f func(ctx context.Context) (*A, error)) Async[A] { |
There was a problem hiding this comment.
why use a pointer as a result ?
There was a problem hiding this comment.
Great question! I actually would prefer not to! I made this choice because it conforms to many go idioms we already have. We could do just (A, error) and provide the zero value. This of course still allows pointers as you can always let A = *B. I think we can make that change.
There was a problem hiding this comment.
Yes I prefer to keep the use of pointer minimal, learnt from you 😅
ProofOfKeags
added
the
incomplete
|
@Roasbeef: review reminder |
|
Closing due to inactivity |
4 similar comments
|
Closing due to inactivity |
|
Closing due to inactivity |
|
Closing due to inactivity |
|
Closing due to inactivity |
|
Closing due to inactivity |
13 similar comments
|
Closing due to inactivity |
|
Closing due to inactivity |
|
Closing due to inactivity |
|
Closing due to inactivity |
|
Closing due to inactivity |
|
Closing due to inactivity |
|
Closing due to inactivity |
|
Closing due to inactivity |
|
Closing due to inactivity |
|
Closing due to inactivity |
|
Closing due to inactivity |
|
Closing due to inactivity |
|
Closing due to inactivity |
|
closing as per bot instructions - feel free to open again once activity continues! |
Change Description
This is an attempt to give us some better building blocks for controlling threads of execution. It was inspired by this PR review.
This PR is intentionally submitted for review despite being incomplete so that it can seek concept/approach ACKs. If we like it. I'll clean up the docs and go through the itests and change them to use this API.
Steps to Test
Steps for reviewers to follow to test the change.
Pull Request Checklist
Testing
Code Style and Documentation
[skip ci]in the commit message for small changes.📝 Please see our Contribution Guidelines for further guidance.