dask-distributed-jl.jl

using MsgPack

function send_msg(io, msg)
    header = Dict()
    messages = [header, msg]
    frames = [MsgPack.pack(msg) for msg in messages]

    write(io, convert(UInt64, length(frames)))
    for frame in frames
        write(io, convert(UInt64, length(frame)))
    end
    for frame in frames
        write(io, frame)
    end

    # 8 bytes -> # of frames
    # N * 8 bytes -> lengths
    # blobs (1st one is MsgPack - specific to the protocol)
end

function read_int(io)
    buf = read(io, 8)
    reinterpret(UInt64, buf)[1]
end

function recv_msg(io)
    N = read_int(io)
    lengths = [read_int(io) for i in 1:N]
    frames = [read(io, l) for l in lengths]
    header, msg = map(x->!isempty(x) ? MsgPack.unpack(x) : Dict(), frames)
    msg
end