# UDP receive and send with julia

**URL:** <https://discourse.julialang.org/t/udp-receive-and-send-with-julia/62349>\
**Category:** New to Julia\
**Created:** [June 3, 2021, 7:03pm UTC](https://discourse.julialang.org/t/udp-receive-and-send-with-julia/62349 "2021-06-03T19:03:53Z")\
**Posts on this page:** 18\
**Page:** 1

<div class="post-metadata">

**Author:** ![Beh123](https://avatars.discourse-cdn.com/v4/letter/b/13edae/32.png) [@Beh123](https://discourse.julialang.org/u/Beh123)\
**Post date:** [June 3, 2021, 7:03pm UTC](https://discourse.julialang.org/t/udp-receive-and-send-with-julia/62349/1 "2021-06-03T19:03:54Z")

</div>

hi. i want to write a program that receive a lot of data with Udp. next i do some operations on data in a function. finally, i send my results to another udp. these operations must be done while i receive data. can i do these work in julia?  
the sample code is here.  
using Sockets  
function recdata()  
@async begin  
sock=UDPSocket()  
bind(sock,ip"x.x.x.x",port)  
while true  
mydata=reinterpret(Int8,recv(sock))  
func1(mydata)  
end  
close(sock)  
end  
end  
function func1(data)  
####some operation that take maximum 1 second  
senddata(output);  
end  
function senddata(output)  
data=reinterpret(Int8,output)  
soc=UDPSocket()  
send(soc, ip"y.y.y.y",port2,data)  
sleep(0.2)  
end

i think, in this method i have lost data. i can do this work with threads in other programs like c and java and etc. commonly, i defined several threads for receiving and processing and sending data.  
can i do similar work in julia? i receive best performance with julia in processing time and need use julia.

---

<div class="post-metadata">

**Author:** ![pbayer](https://sea2.discourse-cdn.com/julialang/user_avatar/discourse.julialang.org/pbayer/32/11675_2.png) [@pbayer](https://discourse.julialang.org/u/pbayer)\
**Post date:** [June 4, 2021, 9:40am UTC](https://discourse.julialang.org/t/udp-receive-and-send-with-julia/62349/2 "2021-06-04T09:40:45Z")

</div>

> [@Beh123](#):
>
> ```julia
> using Sockets
> function recdata()
> @async begin
> sock=UDPSocket()
> bind(sock,ip"x.x.x.x",port)
> while true
> mydata=reinterpret(Int8,recv(sock)) # ??
> func1(mydata)
> end ## this loop will never terminate
> close(sock)
> end
> end
> 
> function func1(data)
> ####some operation that take maximum 1 second
> senddata(output);
> end
> function senddata(output)
> data=reinterpret(Int8,output) ## ???
> soc=UDPSocket()
> send(soc, ip"y.y.y.y",port2,data)
> sleep(0.2)
> end
> 
> ```

I reformatted your code using triple backticks (see [Please read: make it easier to help you](https://discourse.julialang.org/t/psa-make-it-easier-to-help-you/14757)).

Normally you will use some [serialization](https://docs.julialang.org/en/v1.6/stdlib/Serialization/) (`serialize` and `deserialize`) to transmit data via an UDP socket (See also [GitHub - JuliaIO/JSON.jl: JSON parsing and printing](https://github.com/JuliaIO/JSON.jl)). Can `reinterpret` work in your case? Transmission over UDP is done in `UINT8`.

---

<div class="post-metadata">

**Author:** ![tt1234567](https://avatars.discourse-cdn.com/v4/letter/t/dec6dc/32.png) [@tt1234567](https://discourse.julialang.org/u/tt1234567)\
**Post date:** [June 4, 2021, 1:52pm UTC](https://discourse.julialang.org/t/udp-receive-and-send-with-julia/62349/3 "2021-06-04T13:52:11Z")

</div>

I have an application which reads UDP data, processes it and displays it. I don’t know if that’s similar to your case, but if it is then the following might be useful:

- I separated the udp receive, processing and display into separate tasks running on different threads using `ThreadPools.@tspawnat`. For some reason `Threads.@spawn` assigns tasks to random threads and they can end up all running on the same one.

- I used `Channel()` for a blocking, thread-safe queue to communicate between tasks. This allows the receive to carry on if, e.g. the processing operates on larger blocks and could cause packets to be dropped.

- I found I had to increase the UDP buffer size as the default kept dropping packets. You may not have to do that depending on data rate, etc. - mine runs on a Windows laptop and receives around 65 Mbps. I don’t think `UDPSocket()` allows you to change the buffer size directly but there’s a dicussion on how to do it at [Set UDPSocket reciever buffer size? - #4 by ufechner7](https://discourse.julialang.org/t/set-udpsocket-reciever-buffer-size/3939/4)

---

<div class="post-metadata">

**Author:** ![Beh123](https://avatars.discourse-cdn.com/v4/letter/b/13edae/32.png) [@Beh123](https://discourse.julialang.org/u/Beh123)\
**Post date:** [June 4, 2021, 2:30pm UTC](https://discourse.julialang.org/t/udp-receive-and-send-with-julia/62349/4 "2021-06-04T14:30:27Z")

</div>

Thanks a lot. Can you explain tasks with an example.

---

<div class="post-metadata">

**Author:** ![tt1234567](https://avatars.discourse-cdn.com/v4/letter/t/dec6dc/32.png) [@tt1234567](https://discourse.julialang.org/u/tt1234567)\
**Post date:** [June 4, 2021, 3:03pm UTC](https://discourse.julialang.org/t/udp-receive-and-send-with-julia/62349/5 "2021-06-04T15:03:32Z")

</div>

There’s lots in the documentation about tasks, threads, etc. I’m only a beginner so there may well be better ways to do all this, but (making sure you’re running Julia with multiple threads) try

```julia
using ThreadPools

function f1()
    println("f1 in thread ", Threads.threadid())
end

task = ThreadPools.@tspawnat 2 f1()

```

---

<div class="post-metadata">

**Author:** ![pbayer](https://sea2.discourse-cdn.com/julialang/user_avatar/discourse.julialang.org/pbayer/32/11675_2.png) [@pbayer](https://discourse.julialang.org/u/pbayer)\
**Post date:** [June 5, 2021, 6:15am UTC](https://discourse.julialang.org/t/udp-receive-and-send-with-julia/62349/6 "2021-06-05T06:15:38Z")

</div>

Sorry that I didn’t notice it in my first response: you implemented a **sequential** server task. If you get more UDP messages than your task can handle, you will loose packages, because your `recdata` task is not responsive.

The right strategy then is to switch to a **parallel** server: You `@spawn` a parallel server task each time you receive a message and thus keep the `rec_data` task responsive to new messages. The pattern is as follows:

```julia
using Sockets

function rec_data(port::Integer, to::Sockets.InetAddr)
    sock=UDPSocket()
    if bind(sock, Sockets.localhost, port)
        Threads.@spawn begin
            while true
                from, pck = recvfrom(sock)
                msg = deserialize(pck)
                if msg == :exit
                    break
                else
                    Threads.@spawn par_serve(from, to, msg)
                end
            end
            close(sock)
        end
    else
        println(stderr, "port $port not available")
    end
end

function par_serve(from, to, msg)
    # do something with msg and from
    output = some_op_taking_time(from, msg)
    outpck = serialize(output)
    sock = UDPSocket()
    send(sock, to.host, to.port, outpck)
    close(sock)
end

```

This should be sufficient for most situations. If it is not, you must increase the bandwidth of your UDP connections and serve on more ports.

Please notice that this may create a new bottleneck in your application, namely in the server you send your output to. You must handle that in a similar way or perhaps you can use a `Channel` for that as mentioned above.

---

<div class="post-metadata">

**Author:** ![tt1234567](https://avatars.discourse-cdn.com/v4/letter/t/dec6dc/32.png) [@tt1234567](https://discourse.julialang.org/u/tt1234567)\
**Post date:** [June 5, 2021, 12:36pm UTC](https://discourse.julialang.org/t/udp-receive-and-send-with-julia/62349/7 "2021-06-05T12:36:47Z")

</div>

Probably a minor point, but `Threads.@spawn` didn’t work very well for me as I couldn’t find any way of controlling which thread the tasks ran on. If the receive task was scheduled on the same thread as the processing/display, which hapened quite often, it ended up dropping packets. It may well be that my understanding is lacking but, as I mentioned above, I used `ThreadPools` instead and that seems to work OK.

---

<div class="post-metadata">

**Author:** ![Beh123](https://avatars.discourse-cdn.com/v4/letter/b/13edae/32.png) [@Beh123](https://discourse.julialang.org/u/Beh123)\
**Post date:** [June 5, 2021, 6:28pm UTC](https://discourse.julialang.org/t/udp-receive-and-send-with-julia/62349/8 "2021-06-05T18:28:11Z")

</div>

Thanks a lot.  
My receive packet rate is very high. It’s almost. 60Mbps. I have to receive data then process them. My processes are like as multiplication and etc. My process time is one second then I must send 50Mbps data in other udp network.

---

<div class="post-metadata">

**Author:** ![pbayer](https://sea2.discourse-cdn.com/julialang/user_avatar/discourse.julialang.org/pbayer/32/11675_2.png) [@pbayer](https://discourse.julialang.org/u/pbayer)\
**Post date:** [June 5, 2021, 7:43pm UTC](https://discourse.julialang.org/t/udp-receive-and-send-with-julia/62349/9 "2021-06-05T19:43:01Z")

</div>

> [@tt1234567](#):
>
> `Threads.@spawn` didn’t work very well for me as I couldn’t find any way of controlling which thread the tasks ran on. If the receive task was scheduled on the same thread as the processing/display, which hapened quite often, it ended up dropping packets. … I used `ThreadPools` instead

Yes, I can imagine circumstances where having the receiving task on a dedicated thread may be necessary. The Julia scheduler is not preemptive and thus a serving task running o the same thread as the receiving task may block the latter for some time. It depends mainly on the service time and the rate of incoming messages. I wonder if we could do some measurements on that.

> [@Beh123](#):
>
> My processes are like as multiplication and etc. My process time is one second then I must send 50Mbps data in other udp network.

Is this your maximum service time? If yes, your real computation/service times should be much lower.

---

<div class="post-metadata">

**Author:** ![tt1234567](https://avatars.discourse-cdn.com/v4/letter/t/dec6dc/32.png) [@tt1234567](https://discourse.julialang.org/u/tt1234567)\
**Post date:** [June 5, 2021, 7:55pm UTC](https://discourse.julialang.org/t/udp-receive-and-send-with-julia/62349/10 "2021-06-05T19:55:50Z")

</div>

I confess I’ve got a bit lost now. In the code above, each packet seems to be processed and sent separately. If the processing of each packet takes around 1 s I don’t see how you’ll be able to keep up with the 60 Mbps input unless you’ve got a lot of parallel computing power available or the packets are extremely large. This isn’t really my area of expertise so I’m probably missing something.

---

<div class="post-metadata">

**Author:** ![Beh123](https://avatars.discourse-cdn.com/v4/letter/b/13edae/32.png) [@Beh123](https://discourse.julialang.org/u/Beh123)\
**Post date:** [June 6, 2021, 12:52am UTC](https://discourse.julialang.org/t/udp-receive-and-send-with-julia/62349/11 "2021-06-06T00:52:06Z")

</div>

I must explain clearly. I wrote processing and send function. But I didn’t add receiving function. I made rand and etc for making input. My input sizes are 2 matrices that have a lot of complex values. I process them and then send result. My main question is adding receiving function that add data to circular buffer. When size queue is specific value like as 10000 complex number. I must start processing them. When Each 10000 complex number added circular queue , I must start process and send function. My input package are extremely large. Can I add large packet in circular buffer and process them when the capacity of buffer is 10000?  
We have receive( msg, nbytes) method in vc++ for reading n bytes on network. Is there similar method in julia for reading n bytes in julia?

---

<div class="post-metadata">

**Author:** ![pbayer](https://sea2.discourse-cdn.com/julialang/user_avatar/discourse.julialang.org/pbayer/32/11675_2.png) [@pbayer](https://discourse.julialang.org/u/pbayer)\
**Post date:** [June 6, 2021, 9:13am UTC](https://discourse.julialang.org/t/udp-receive-and-send-with-julia/62349/12 "2021-06-06T09:13:14Z")

</div>

Please provide an MWE as described [here](https://discourse.julialang.org/t/psa-make-it-easier-to-help-you/14757) since I’m lost with your explanation too.

> [@Beh123](#):
>
> Is there similar method in julia for reading n bytes in julia?

`n` is given by the UDP package size received with `pck = recv(sock)`.

---

<div class="post-metadata">

**Author:** ![tt1234567](https://avatars.discourse-cdn.com/v4/letter/t/dec6dc/32.png) [@tt1234567](https://discourse.julialang.org/u/tt1234567)\
**Post date:** [June 6, 2021, 9:36am UTC](https://discourse.julialang.org/t/udp-receive-and-send-with-julia/62349/13 "2021-06-06T09:36:26Z")

</div>

Yes, a MWE would be helpful but I’m guessing that you might want a structure a bit like the following (modifying @pbayer’s code a bit). It’s close to what I do but I’m a beginner and there may be better ways of doing, it in which case I’d love to know how to improve it.

```julia
using Sockets
using ThreadPools

function rec_data(port::Integer, to::Sockets.InetAddr, chan::Channel)
    sock=UDPSocket()
    if bind(sock, Sockets.localhost, port)
        Threads.@spawn begin
            while true
                from, pck = recvfrom(sock)
                put!(chan, pck
                if msg == :exit
                    break
                else
                    Threads.@spawn par_serve(from, to, msg)
                end
            end
            close(sock)
        end
    else
        println(stderr, "port $port not available")
    end
end

function process(chan::Channel)
    while some_condition
        packet = take!(chan)
        # convert packet contents to complex numbers or whatever
        # put data into suitable data structure (circular buffer/array/etc)
        if enough_data
            do_complicated_processing()
            send_data()
        end
    end
end

queuelength = 8000 # or something long enough
chan = Channel{Array{UInt8,1}}(queuelength)
rectask = ThreadPools.@tspawnat 2 rec_data(port, to, chan)
proctask = ThreadPools.@tspawnat 3 process(chan)

```

---

<div class="post-metadata">

**Author:** ![tt1234567](https://avatars.discourse-cdn.com/v4/letter/t/dec6dc/32.png) [@tt1234567](https://discourse.julialang.org/u/tt1234567)\
**Post date:** [June 6, 2021, 9:42am UTC](https://discourse.julialang.org/t/udp-receive-and-send-with-julia/62349/14 "2021-06-06T09:42:22Z")

</div>

> Yes, I can imagine circumstances where having the receiving task on a dedicated thread may be necessary. The Julia scheduler is not preemptive and thus a serving task running o the same thread as the receiving task may block the latter for some time. It depends mainly on the service time and the rate of incoming messages. I wonder if we could do some measurements on that.

Some means of stopping a high-priority task from being blocked would help my application. At the moment my processing is fine on a single thread but in the next stage of development I’ll want to parallelise it.

---

<div class="post-metadata">

**Author:** ![pbayer](https://sea2.discourse-cdn.com/julialang/user_avatar/discourse.julialang.org/pbayer/32/11675_2.png) [@pbayer](https://discourse.julialang.org/u/pbayer)\
**Post date:** [June 6, 2021, 11:33am UTC](https://discourse.julialang.org/t/udp-receive-and-send-with-julia/62349/15 "2021-06-06T11:33:14Z")

</div>

> [@tt1234567](#):
>
> At the moment my processing is fine on a single thread but in the next stage of development I’ll want to parallelise it.

With your proposal above you are halfway to parallelization. You still have a bottleneck, namely in the `do_complicated_processing()`, which is done serially in your single processing task.

But you can change that to a broker task like the following:

```julia
using Sockets
using ThreadPools

function rec_data(port::Integer, to::Sockets.InetAddr, chan::Channel)
    sock=UDPSocket()
    if bind(sock, Sockets.localhost, port)
        Threads.@spawn begin
            while true
                from, pck = recvfrom(sock)
                msg = serialize(pkg)
                if msg == :exit
                    break
                else
                    put!(chan, msg) # send it to the broker
                end
            end
            close(sock)
        end
    else
        println(stderr, "port $port not available")
    end
end

function broker(chan::Channel)
    while some_condition
        msg = take!(chan)
        # convert message contents to complex numbers or whatever
        # put data into suitable data structure (circular buffer/array/etc)
        if enough(data)
            ThreadPools.@tspawnat rand(4:Threads.nthreads()) begin   
                do_complicated_processing_and_send(data)
            end
        end
    end
end

queuelength = 8000 # or something long enough
chan = Channel{Array{UInt8,1}}(queuelength)
rectask = ThreadPools.@tspawnat 2 rec_data(port, to, chan)
brokertask = ThreadPools.@tspawnat 3 broker(chan)

```

The broker task is responsive and starts a parallel task on some thread \>= 4 if enough data is available.

> [@tt1234567](#):
>
> Some means of stopping a high-priority task from being blocked would help my application.

I think this is not possible at the moment, but the outlined strategy should do. Notice that a task yields to the scheduler (blocks) when it does a `recv` or `take!`. The problem is that it gets not scheduled again until the subsequent task on that thread has finished its work or yields. Therefore it makes sense to keep at least the `rec_data` task on a separate thread to stay responsive to incoming UDP messages.

I further think that the system would benefit if the data chunks going to the complicated processing don’t get too big. If the processing is shorter and done more often, the system will parallelize better.

---

<div class="post-metadata">

**Author:** ![tt1234567](https://avatars.discourse-cdn.com/v4/letter/t/dec6dc/32.png) [@tt1234567](https://discourse.julialang.org/u/tt1234567)\
**Post date:** [June 6, 2021, 12:10pm UTC](https://discourse.julialang.org/t/udp-receive-and-send-with-julia/62349/16 "2021-06-06T12:10:52Z")

</div>

Thanks, it sounds as though I’m not too far off the right lines and hopefully it will cover the OP’s case as well. Parallelising the processing obviously means the output ordering can’t be guaranteed so, depending on the application, some marshalling might be needed before sending.  
My instincts from (limited) experience in parallel processing in C++ would be to have a number of fixed processing tasks and distribute the data vis queues of some sort. Does `@tspwanat` have sufficiently low overhead to make that unnecessary?

---

<div class="post-metadata">

**Author:** ![Beh123](https://avatars.discourse-cdn.com/v4/letter/b/13edae/32.png) [@Beh123](https://discourse.julialang.org/u/Beh123)\
**Post date:** [June 14, 2021, 8:49am UTC](https://discourse.julialang.org/t/udp-receive-and-send-with-julia/62349/17 "2021-06-14T08:49:37Z")

</div>

Thanks a lot.  
When I use rec\_ data function. I see Port not available.  
I have another program that send data. But my program shows port 2000 not avaliable all time.

---

<div class="post-metadata">

**Author:** ![pbayer](https://sea2.discourse-cdn.com/julialang/user_avatar/discourse.julialang.org/pbayer/32/11675_2.png) [@pbayer](https://discourse.julialang.org/u/pbayer)\
**Post date:** [June 14, 2021, 10:46am UTC](https://discourse.julialang.org/t/udp-receive-and-send-with-julia/62349/18 "2021-06-14T10:46:52Z")

</div>

> [@Beh123](#):
>
> When I use rec\_ data function. I see Port not available. … my program shows port 2000 not avaliable all time.

This means that port 2000 is already used by another socket and `bind(sock, Sockets.localhost, 2000)` returned `false`.

If you have the socket variable (bound to 2000) available, you can

- `close` it to allow for a new socket to `bind` to that port or
- modify `rec_data` accordingly to receive over that socket.

If you bound a socket unwittingly to that port, you can close your Julia session to free that port again.

Or (if that port is used by another program) you must use another port.