optimized send - direct writes for large bitstype arrays

[amitmurthy]

Whenever a request exceeds 100K in serialized length, it is directly written to the socket.

Supersedes #6876, partially address #9992

[amitmurthy]

@andreasnoack :

Current master:

julia> a=ones(10^7);

julia> println(mean([@elapsed remotecall_fetch(2, (x)->x, a) for i in 1:10]))
0.1583344692

julia> s=[randstring() for x in 1:10^5];

julia> println(mean([@elapsed remotecall_fetch(2, (x)->x, s) for i in 1:10]))
0.2919245722

julia> println(@elapsed [remotecall_fetch(2, myid) for i in 1:10000])
0.618292734

With this PR:

julia> a=ones(10^7);

julia> println(mean([@elapsed remotecall_fetch(2, (x)->x, a) for i in 1:10]))
0.0611091225

julia> s=[randstring() for x in 1:10^5];

julia> println(mean([@elapsed remotecall_fetch(2, (x)->x, s) for i in 1:10]))
0.1801866141

julia> println(@elapsed [remotecall_fetch(2, myid) for i in 1:10000])
0.602078664

[amitmurthy]

@andreasnoack - could you please test this out? I would like to merge it soon.
@vtjnash / @Keno - could you review the code if possible?

[andreasnoack]

Yes. Sorry for not getting back to you on this one. I'll run the benchmarks tomorrow and report back.

[andreasnoack]

Her are the results and they confirm your findings. This is really great. First the plot of the time it takes to move an array of Float64s to a worker against the size of the array.

and next the relative timings to MPI with network transport (MPI-TCP).

So instead of being four times MPI we'll now be two times MPI (for large arrays).
cc: @alanedelman

[ViralBShah]

Do we know where we are losing the 2x now?

If we can make this work as well as MPI-tcp, that may lead to reconsidering the darray design. Of course, we would also need efficient implementation of collectives at some basic level.

[amitmurthy]

Not yet, but we will keep chipping away. The IOBuffer used to aggregate messages till 100K in length can be replaced by a fixed length array - will remove unnecessary allocs for small requests too.

Optimizing @everywhere where we serialize the same request multiple times (and send it multiple times too) is another TODO.

[JeffBezanson]

This is great! Could you briefly explain the reason to introduce BufferedAsyncStream rather than add this behavior to existing AsyncStream types?

[amitmurthy]

We should add it to all existing AsyncStream types. In #6876 there was a discussion about having this support for all IO, adding a buffer to File and getting rid of IOStream entirely - which I am unfamiliar on how to go about.

I figured it best to redo the PR to just solve the single problem for sending large arrays in our parallel infrastructure while the other issues get sorted out.

I can submit a PR that will move the implementation to all AsyncStream types, while exporting 3 new functions - lock, unlock and set_lockable. The last one since by default we want AsyncStream's to retain existing behavior, i.e., no buffering.