Universal API for use of closeable resources with coroutines

[elizarov]

For background of the problem see #1044. There we've introduced a special API for CancellableContinuation (resume(r) { r.close() }) so that you can resume with a "closeable resource" that would be atomically close if the continuation is cancelled. However, there are other APIs in coroutines that can potentially lose closeable resource on cancellation:

• async { ... } may produce a result but then get cancelled in a race, so that the resulting resource is lost (not closed)
• withContext/withTimeout/coroutineScope/etc may return a resource, but get cancelled on return path.
• Resources can be send to channel, and when a channel is cancelled all the buffered resources a lost.
• Channels current have "atomic send/recieve" guarantee that one may want to chance to cancellable operations, but then resources will get lost.

So the proposal is to add a universal API for closeable resources in form of a wrapper class that is used like this: Resource(v) { v.close() }

So, for example, that if how you can use it with async:

val deferred: Deferred<Resource<V>> = async { 
    val v = openSomeResource() 
    return@async Resource(v) { v.close() }
}

Now if deferred is cancelled, the resource is closed. You can use it like:

deferred.await().value // get value from Resource

Additional considerations

1. We may also want to introduce some interface with close or dispose function that does the similar trick for classes implementing this interface (and make Resource implement it). The name of this interface is hard to come by with, though.

2. Maybe this is a wrong approach altogether. Maybe we shall look at scoped resource management.

[qwwdfsad]

More considerations:

• Using Resource<V> may be too intrusive for users and their API and for kotlinx.coroutines maintenance
• Alternative approach is to provide closeOnCancel alternative:

withContext(IO) {
    val resource = ...
    resource.closeOnCancel { resource.close() } // Bikeshed naming
    resource 
}

where closeOnCancel is implemented as suspend function and invokeOnCancellation

[elizarov]

closeCancel looks nice for withContext and Context, but does not work for channels. However, we can extract some common "cancellable scope" interface that both coroutine scope and channels would implement and make it possible in such a scope to write:

withContext(IO) { 
    val resource = ...
    invokeOnCancel { resource.close() } // extension on CoroutineScope
    resource 
}

Then the same API for channels:

    val resource = ...
    channel.invokeOnCancel { resource.close() } 
    channel.send(resource)

where a shortcut can be also made:

    val resource = ...
    channel.send(resource) { resource.close() }

[fvasco]

@elizarov Can channel.invokeOnCancel work?
channel must dispose the resource if the send is ok and the receive fails, otherwise the resource must keep alive though the channel was cancelled.

[elizarov]

@fvasco Good catch. channel.invokeOnCancel does not seem to be fruitful idea. It has to be either a wrapper like channel.send(Resource(r) { r.close() }), a special method like channel.send(r) { r.close() } or a special strategy that is attached to the channel and configures closing of all resources in it.

[fvasco]

@elizarov

a special method like channel.send(r) { r.close() }

Special method can works for Channel, but I'm curious about the withContext's syntax.
A downside is have specify the disposer on each send invocation, same for onSends.
Another trivial consideration is: the disposer should be context free (possibly a singleton), ie:

typealias Disposer<T> =suspend (T) -> Unit

channel.send(r, Resource::close)

These consideration should be applicable to Flow too, should filter or drop operators dispose the discarded elements?

Should this behaviour be a part of CoroutineContext, like CoroutineExceptionHandler?

interface DisposeHandler : CoroutineContext.Element {

    /**
     * Dispose [resource] and returns `true`, `false` otherwise
     */
    suspend fun dispose(resource: Any): Boolean
}

object CloseableDisposeHandler : DisposeHandler, CoroutineContext.Key<CloseableDisposeHandler> {

    override suspend fun dispose(resource: Any): Boolean {
        return if (resource is Closeable) {
            // the Dispatcher should be IO, the same of withContext
            resource.close()
            true
        } else false
    }

    override val key: CoroutineContext.Key<*> get() = CloseableDisposeHandler
}

suspend fun main() {
    coroutineScope {
        val tempFile = withContext(Dispatchers.IO + CloseableDisposeHandler) {
            createTempFile()
        }
    }
}

Should CoroutineContext contains multiple disposers?

withContext(Resource1Disposer + Resource2Disposer) {
    val r1 = async { loadResource1() }
    val r2 = async { loadResource2() }
}

[elizarov]

Adding disposers to the context and then selecting them based on the runtime "instanceof" checks seems to be a dangerous road to be.

For files, in particular, a dedicated scoped API seems to work better:

suspend fun main() {
    disposeScope { // will close all open files, purely speculative name
        coroutineScope {
            val tempFile = withContext(Dispatchers.IO) {
                 createTempFile() // extension on DisposeScope
            }
        }
    }
}

So this solution is othogonal to coroutines and works great with withContext { ... }, but does not readily work with channels.

[LouisCAD]

Would there still be no guarantee as to where (thread speaking) the close lambda would be executed, and would it still be a blocking lambda, with no suspension points support?

[elizarov]

After some discussion, we've decided that "universal" wrapper API like a Resource proposed here is both quite ambitious in its "universality" and quite inconvenient to use. So I'm retracting this proposal and closing this issue. On a side note, it would not have solved for cases when some 3rd party code creates a wrapper around a reference and returns it from a cancellable coroutine anyway.

In light of #1813 we decided for a more modest replacement. See #1936.