Arvados

10467-client-disconnect -- test 9411197dfa8ff4c7d935a395a04b5846c7b52ffd

(note -- this is based on a not-yet-merged 10468 branch)

• Uses context.TODO() but documentation says to use context.Background() (https://golang.org/pkg/context/)
• Unix and Azure blob volume doesn't support canceling Get() (future story)
• I feel like we would benefit from an additional abstraction here, something like a "cancellable pipe" which implements the behavior of passing data through or on cancel abruptly terminates the current read or write. This would make cancellation simpler because by exposing it as as IO read or write errors.
• I don't see any tests that cancel anything?

Peter Amstutz wrote:

• Uses context.TODO() but documentation says to use context.Background() (https://golang.org/pkg/context/)

Ah yes, fixed tests and pull_worker. Handlers still use TODO "because the surrounding function has not yet been extended to accept a Context parameter".

• I feel like we would benefit from an additional abstraction here, something like a "cancellable pipe" which implements the behavior of passing data through or on cancel abruptly terminates the current read or write. This would make cancellation simpler because by exposing it as as IO read or write errors.

I think what you're looking for is io.Pipe() (see "CloseWithError") which I used in s3 PUT to ensure a stalled s3 operation can't wake up later and resume use of a buffer that has long since been repurposed for a different request.

We could move that up a layer if we refactored our Volume interface to use readers and writers instead of passing a buffer. Is that what you mean? (It might be tricky to implement the "concurrent GET requests" Azure volume feature without using 2x memory, but then again, maybe that concurrency feature should also be moved up a layer anyway, and volumes should just support range requests.)

• I don't see any tests that cancel anything?

Yeah, that'd be good.

Added tests for Get, Put, Compare.

10467-client-disconnect - test e675118bd2b28ec40833d06ea384b6f1c78f3039

Tests LGTM.

Azure implementation on 10467-client-disconnect @ d0dc273718a2f657643a2b2800d984c7a3a62f78

Tom Clegg wrote:

Azure implementation on 10467-client-disconnect @ d0dc273718a2f657643a2b2800d984c7a3a62f78

This LGTM.

10467-client-disconnect @ 568c7abf660b7a68f70b6ea47ae2e7352233f053 https://ci.curoverse.com/job/developer-run-tests/121/

If this goes well, next the Azure and S3 drivers can be refactored to use these getWithPipe() and putWithPipe() funcs. That should simplify them, and pave the way to changing the Volume interface to use io.PipeReader and io.PipeWriter instead of passing bufs around.

The approach of passing functions (getWithPipe, get, getFunc) into other functions in order to wrap the behavior is hard to follow and feels a bit like Ruby written in Go. I think this would be a bit more Go-like to return an io.Reader and have the Close() method take care of calling unlock(), similar to errorReaderCloser. E.g.

type volumeUnlockReader struct {
    *io.ReaderCloser
    v *UnixVolume
}

func (rw *volumeUnlockReader) Close() err {
    defer rw.v.unlock()
    return rw.Reader.Close()
}

func (v *UnixVolume) getBlockReader(ctx context.Context, path string) (io.Reader, error) {
    if err := v.lock(ctx); err != nil {
        return nil, err
    }
    f, err := os.Open(path)
    if err != nil {
        v.unlock()
        return nil, err
    }
    return &ReaderWrapper{f, v}, nil
}

func (v *UnixVolume) get(ctx context.Context, loc string, w *io.PipeWriter) {
    path := v.blockPath(loc)
    stat, err := v.stat(path)
    if err != nil {
        w.CloseWithError(v.translateError(err))
        return
    }
    rdr, err = v.getBlockReader(ctx, path)
    if rdr != nil {
        defer rdr.Close()
    }
    if err != nil {
        return
    }
    _, err = io.CopyN(w, rdr, n)
    w.CloseWithError(err)
}

If I'm following, the road map here is:

1. Convert all the volumes to use getWithPipe/putWithPipe
2. Change the volume interface to Get(ctx context.Context, loc string, w *io.PipeWriter) and Put(ctx context.Context, loc string, rdr io.ReadCloser) and invoke getWithPipe/putWithPipe in the handler
3. Change getWithPipe/putWithPipe to take a target Writer or Reader instead of a buffer so that we can stream directly to/from the socket.

Is that about right?

Peter Amstutz wrote:

The approach of passing functions (getWithPipe, get, getFunc) into other functions in order to wrap the behavior is hard to follow

Yeah, the focus is on making it easy to write/maintain N "getter" functions.

When all volumes have getters (or a Get() that returns a WriterTo) this glue can move up the stack and won't have to adapt one interface to another any more.

If I'm following, the road map here is:

1. Convert all the volumes to use getWithPipe/putWithPipe
2. Change the volume interface to Get(ctx context.Context, loc string, w *io.PipeWriter) and Put(ctx context.Context, loc string, rdr io.ReadCloser) and invoke getWithPipe/putWithPipe in the handler
3. Change getWithPipe/putWithPipe to take a target Writer or Reader instead of a buffer so that we can stream directly to/from the socket.

Is that about right?

Yes, except getWithPipe/putWithPipe will probably just take a *Volume, and call the appropriate Get/Put method.

Updated to use io.Writer and io.Reader interfaces instead of *io.PipeWriter and *io.PipeReader, so getter/putter funcs are more like io.WriterTo and io.ReaderFrom.

Removed some debug printf.

b4d9dfe1e7acb1f45c2cc699020bf9299a0db5c9

... 3f556ca1b44b7e01874bd172abbb7cb3df0615db with BlockWriter and BlockReader interfaces instead of passing funcs around.

LGTM @ 3f556ca1b44b7e01874bd172abbb7cb3df0615db