tcpsocket supports a send buffer

[amitmurthy]

In the same spirit as #6768, this PR avoids a copy while transmitting large arrays between workers. It does this by

1. moving the buffering functionality of worker connections to becoming a feature of tcpsocket itself and
2. directly writing large arrays to the socket instead of via an intermediate buffer.

[hisq]

Probably also fixes #4508.

[JeffBezanson]

Awesome!! Looks pretty good but I will read it in more detail later.

[vtjnash]

would it make sense to make this a feature of all our streams, add generic lock/unlock functions, and utilize them whenever entering/exiting IO-related functions (e.g. print/show)?

[Keno]

I agree with Jameson. This looks fine to me and I agree that we shouldn't by default buffer just a single write call, since that would be confusing.

[amitmurthy]

@vtjnash did you mean all streams (including IOStream) or only AsyncStreams?

[vtjnash]

perhaps only AsyncStreams, but all streams would be even nicer. it doesn't entirely makes sense for IOBuffer to be pre-buffered, but it does make sense for it to have a write lock.

This would make IOStream double-buffered, but we could then replace IOStream with uv_file in general and get non-blocking IO for everything (which could potentially make the Pkg manager faster, since it appears to be I/O limited doing a independent operations on metadata for a lot of files).

[JeffBezanson]

Yes, with a buffer on the libuv File we can get rid of IOStream entirely (assuming it will perform well).

[amitmurthy]

Currently using a RemoteRef for locking. Wondering if I should change to use libuv mutexes. Local remote refs in workers suffer from the fact that they cannot be initialized till the worker id is known. Also libuv mutexes should be faster and more lightweight.

[amitmurthy]

Just realized that libuv mutexes are recursive and Julia tasks would show up as being part of the same thread. May not be usable.

[amitmurthy]

Have moved the implementation to support all AsyncStreams (except UPDSocket).

[amitmurthy]

No tidy way to solve #3787 without locks though. With locking (if we decide to export the same), user code could look like

Base.make_lockable(STDOUT)
@sync begin
    for i in 1:3
        @async begin
            Base.lock_w(STDOUT) 
            println(`Getting garbled`)
            Base.unlock_w(STDOUT) 
        end
    end
end

[amitmurthy]

Bump. As stated it now supports all AsyncStreams. Feedback?

Merging of File and IOStream is a bit more involved and I would like to do that separately.

[Keno]

What's the need for having the lock inside the IOStream rather than as a separate thing?

[amitmurthy]

It is more intuitive to lock a stream and then write multiple times to it. Rather than have user code manage a lock separately. Sort of like lockf / fcntl on an open file fd.

Anyways, the lock in the IOStream object is not initialized till make_locakable is called on the stream. This is required since a local RemoteRef cannot be instantiated properly on a worker till its julia pid is known. And the pid is known only when the worker receives the first initialization message from the master process.

[amitmurthy]

Bump. Would be great if this could make it into 0.3

[andreasnoack]

What is the status here? I get i nice speedup from this on large arrays. See the yellow graph. It shows the time it takes to copy an array of Float64s to another process on my laptop as a function of the array size. The hump seems to be genuine as I have rerun the tests couple of times and the reported times are minima of ten runs. Maybe related to the way the buffer grows?

There also seems to be quite a bit of overhead in the @spawn compared to MPI Send/Recv. Is this tracked in an issue or just considered unavoidable because of the way @spawns work?

[amitmurthy]

It is waiting for a review by @JeffBezanson

The hump comes about since this optimization does not kick in for arrays less than 1MB in bytesize. The value of const SENDBUF_SZ.

I think quite a bit of overhead in @spawn is due to serialization itself. For e.g.

julia> iob=IOBuffer()
julia> a=ones(10^8);

julia> @time remotecall_fetch(2, x->x, a);
elapsed time: 0.925497984 seconds (1769649824 bytes allocated, 5.42% gc time)

julia> @time remotecall_fetch(2, x->x, a);
elapsed time: 0.914441734 seconds (1767317752 bytes allocated, 3.63% gc time)

julia> @time serialize(iob, a);
elapsed time: 0.239022428 seconds (967168640 bytes allocated, 5.84% gc time)

julia> seekstart(iob)

julia> @time serialize(iob, a);
elapsed time: 0.131713773 seconds (112 bytes allocated)

So, a local serialization to a pre-allocated buffer takes 0.13 seconds. If not pre-allocated, it is 0.23 seconds. Again the deserialization/serialization on the remote end will have similar figures.

[amitmurthy]

Closed in favor of #10073