# Thread safe buffer which blocks retrieval when empty?
**URL:** https://discourse.julialang.org/t/thread-safe-buffer-which-blocks-retrieval-when-empty/107352
**Category:** General Usage
**Tags:** question
**Created:** [December 9, 2023, 9:36am UTC](https://discourse.julialang.org/t/thread-safe-buffer-which-blocks-retrieval-when-empty/107352 "2023-12-09T09:36:35Z")
**Posts on this page:** 12
**Page:** 1
### Author:  [@tamasgal](https://discourse.julialang.org/u/tamasgal)
#### Post date: [December 9, 2023, 9:36am UTC](https://discourse.julialang.org/t/thread-safe-buffer-which-blocks-retrieval-when-empty/107352/1 "2023-12-09T09:36:35Z")
Does anyone know of a package which implements a (circular) buffer that blocks when popping from it if it’s empty?
To give a bit more context, below is a MWE for a program which has one thread running and continuously fetching data from a source (with high and sometimes varying rates) and putting it into a circular buffer with a limited size to avoid a memory fill-up. The main thread has an endless loop where it fetches data from this buffer and processes it (which has also varying processing rates).
In the simplified example below, the data rate is 0.5s and the processing rate 1.0s with a buffer which can hold up to 5 elements.
There is some boilerplate code (I know it’s not much) with locks and also an ugly `sleep(0.1)` which is needed, otherwise the loop hangs of course with 100% when the buffer happens to be empty.
Long story short: **is there any package which handles this elegantly and efficiently?** Preferably a package which offers such a buffer type which is thread safe and allows something like `get(buffer; block=true, timeout=...)`?
I looked around but could not find anything yet. Maybe I am overthinking this and the below implementation is already OK as it is 😉
Here is the MWE:
```julia
using DataStructures
DATA_RATE = 0.5
PROCESSING_RATE = 1.0
function datafetcher(buffer, lck)
i = 0
while true
lock(lck)
println(" -> pushing $i to $buffer")
push!(buffer, i)
unlock(lck)
i += 1
sleep(DATA_RATE)
end
end
function main()
buffer = CircularBuffer{Int}(5)
lck = ReentrantLock()
@async datafetcher(buffer, lck)
while true
if isempty(buffer)
sleep(0.1) # needed to not choke on this loop when the buffer is empty
continue
end
lock(lck)
data = popfirst!(buffer)
unlock(lck)
println("processing $data")
sleep(PROCESSING_RATE)
end
end
main()
```
---
### Author:  [@cjdoris](https://discourse.julialang.org/u/cjdoris)
#### Post date: [December 9, 2023, 10:32am UTC](https://discourse.julialang.org/t/thread-safe-buffer-which-blocks-retrieval-when-empty/107352/2 "2023-12-09T10:32:44Z")
Sounds like a classic use for a `Channel`.
---
### Author:  [@tamasgal](https://discourse.julialang.org/u/tamasgal)
#### Post date: [December 9, 2023, 10:50am UTC](https://discourse.julialang.org/t/thread-safe-buffer-which-blocks-retrieval-when-empty/107352/3 "2023-12-09T10:50:12Z")
I was thinking about that actually. I’d need to implement the logic to discard data in the retrieval function.
---
### Author:  [@tamasgal](https://discourse.julialang.org/u/tamasgal)
#### Post date: [December 9, 2023, 7:34pm UTC](https://discourse.julialang.org/t/thread-safe-buffer-which-blocks-retrieval-when-empty/107352/4 "2023-12-09T19:34:53Z")
I guess `Channel` it is, but unfortunately there is no documented way (public API) to check if a channel is full although there is a constructor to create a channel with a given maximum size.
I can of course implement
```julia
isfull(c::Channel) = length(c.data) >= c.sz_max
```
but it doesn’t feel really future proof 😉
```julia
julia> methodswith(Channel)
[1] bind(c::Channel, task::Task) @ Base channels.jl:268
[2] close(c::Channel) @ Base channels.jl:200
[3] close(c::Channel, excp::Exception) @ Base channels.jl:201
[4] fetch(c::Channel) @ Base channels.jl:419
[5] isempty(c::Channel) @ Base channels.jl:534
[6] isopen(c::Channel) @ Base channels.jl:214
[7] isready(c::Channel) @ Base channels.jl:533
[8] iterate(c::Channel) @ Base channels.jl:604
[9] iterate(c::Channel, state) @ Base channels.jl:604
[10] lock(c::Channel) @ Base channels.jl:540
[11] lock(f, c::Channel) @ Base channels.jl:541
[12] put!(c::Channel{T}, v) where T @ Base channels.jl:333
[13] show(io::IO, c::Channel) @ Base channels.jl:585
[14] show(io::IO, ::MIME{Symbol("text/plain")}, c::Channel) @ Base channels.jl:587
[15] take!(c::Channel) @ Base channels.jl:465
[16] trylock(c::Channel) @ Base channels.jl:543
[17] unlock(c::Channel) @ Base channels.jl:542
[18] wait(c::Channel) @ Base channels.jl:569
[19] Distributed.WorkerPool(c::Channel, ref::Distributed.RemoteChannel) @ Distributed ~/.julia/juliaup/julia-1.9.4+0.aarch64.apple.darwin14/share/julia/stdlib/v1.9/Distributed/src/workerpool.jl:26
julia> c = Channel{UInt8}(5)
Channel{UInt8}(5) (empty)
julia> put!(c, 3)
0x03
julia> put!(c, 4)
0x04
julia> put!(c, 5)
0x05
julia> c.data
3-element Vector{UInt8}:
0x03
0x04
0x05
julia> c.
cond_put cond_take cond_wait data excp n_avail_items
state sz_max
julia> c.sz_max
5
```
---
### Author:  [@RomeoV](https://discourse.julialang.org/u/RomeoV)
#### Post date: [December 9, 2023, 9:28pm UTC](https://discourse.julialang.org/t/thread-safe-buffer-which-blocks-retrieval-when-empty/107352/5 "2023-12-09T21:28:35Z")
Why do you need to know when your Channel is full? What would you do in that situation? Discard the result?
---
### Author:  [@RomeoV](https://discourse.julialang.org/u/RomeoV)
#### Post date: [December 9, 2023, 9:34pm UTC](https://discourse.julialang.org/t/thread-safe-buffer-which-blocks-retrieval-when-empty/107352/6 "2023-12-09T21:34:33Z")
Currently `channels.jl` seems to be doing pretty much what your `isfull` function would be doing: [https://github.com/JuliaLang/julia/blob/abeb68feea5da6151a83ec27786a23b71e06fc88/base/channels.jl#L362](https://github.com/JuliaLang/julia/blob/abeb68feea5da6151a83ec27786a23b71e06fc88/base/channels.jl#L362)
```julia
while length(c.data) == c.sz_max
check_channel_state(c)
wait(c.cond_put)
end
```
---
### Author:  [@tamasgal](https://discourse.julialang.org/u/tamasgal)
#### Post date: [December 9, 2023, 9:35pm UTC](https://discourse.julialang.org/t/thread-safe-buffer-which-blocks-retrieval-when-empty/107352/7 "2023-12-09T21:35:05Z")
Yes, I deal with high data rates and when the processing cannot keep up, I need to discard the data to avoid a memory pile-up.
---
### Author:  [@RomeoV](https://discourse.julialang.org/u/RomeoV)
#### Post date: [December 9, 2023, 9:36pm UTC](https://discourse.julialang.org/t/thread-safe-buffer-which-blocks-retrieval-when-empty/107352/8 "2023-12-09T21:36:47Z")
If I were you, I’d probably just copy the function I linked, rename it to `put_buffered_or_discard!`, and then change the while loop to abort if it’s full. That way you can be reasonably sure you don’t put in any bugs with race conditions, assuming that this code doesn’t have any.
---
### Author:  [@tamasgal](https://discourse.julialang.org/u/tamasgal)
#### Post date: [December 9, 2023, 9:38pm UTC](https://discourse.julialang.org/t/thread-safe-buffer-which-blocks-retrieval-when-empty/107352/9 "2023-12-09T21:38:50Z")
Yes, I am preparing a tiny PR with such a function since I think it’s much cleaner to have an exported function than using two fields which are not documented and might change in future `;)`
---
### Author:  [@RomeoV](https://discourse.julialang.org/u/RomeoV)
#### Post date: [December 9, 2023, 9:46pm UTC](https://discourse.julialang.org/t/thread-safe-buffer-which-blocks-retrieval-when-empty/107352/10 "2023-12-09T21:46:09Z")
Sure, in theory they could change in the future – I don’t see why they would though. They’ve been [the same for 8 years](https://github.com/JuliaLang/julia/blob/21ca79976063a56d44a35d786cb95c0e19500861/base/channels.jl#L10-L12).
---
### Author:  [@tamasgal](https://discourse.julialang.org/u/tamasgal)
#### Post date: [December 9, 2023, 9:47pm UTC](https://discourse.julialang.org/t/thread-safe-buffer-which-blocks-retrieval-when-empty/107352/11 "2023-12-09T21:47:53Z")
It’s just good practice 🙂 we’ll see how the PR goes… 😆
---
### Author:  [@danielwe](https://discourse.julialang.org/u/danielwe)
#### Post date: [December 9, 2023, 11:16pm UTC](https://discourse.julialang.org/t/thread-safe-buffer-which-blocks-retrieval-when-empty/107352/12 "2023-12-09T23:16:50Z")
xref this open PR that implements `tryput!`: [add maybetake! and tryput! by tkf · Pull Request #41966 · JuliaLang/julia · GitHub](https://github.com/JuliaLang/julia/pull/41966)
It’s not what you need, as it only gives up if the channel is closed, not if it’s full. Perhaps it’s possible to resurrect this PR and add a keyword arg `tryput!(ch, x; skipfull=true)`?