Repository navigation
sync: use SpinLock for channel - #24802
Merged
Merged
Conversation
|
Connected to Huly®: V_0.6-23216 |
Contributor
|
Can you please run the benchmarks in vlib/sync/bench/ and compare them between master and this PR? |
Contributor
Not sure ... If we do not have a test that explicitly checks, I am inclined to say no. edit: after the examples below, I am convinced that it is correct behavior. |
Contributor
|
I think that reading from a closed channel, that still has buffered items, until they are drained, is useful. import log
import time
import rand
fn writer(ch chan int) {
for {
res := <-ch or { break }
log.warn('${' ':50} saving result: ${res}')
time.sleep(50 * time.millisecond)
}
log.warn('${' ':50} channel was closed, the writer loop broke, writer thread is ending')
}
fn worker(ch chan int, i int) {
log.debug('>> worker ${i} starts')
for _ in 0 .. 2 {
time.sleep((15 + rand.intn(100) or { 0 }) * time.millisecond)
x := rand.intn(10) or { 0 }
value := i * 1000_000 + x
log.debug('>> worker ${i} is sending value: ${value}')
ch <- value
}
log.debug('>> worker ${i} ends')
}
fn main() {
nthreads := arguments()[1] or { '5' }.int()
log.set_level(.debug)
log.info('main start, nthreads: ${nthreads}')
ch := chan int{cap: 10}
consumer := spawn writer(ch)
log.info('consumer started.')
mut producers := []thread{cap: nthreads}
for i in 0 .. nthreads {
producers << spawn worker(ch, i + 1)
}
log.info('workers started.')
producers.wait()
log.info('workers finished.')
ch.close()
log.info('channel closed.')
consumer.wait()
log.info('main done, consumer finished')
} |
Contributor
|
Producing: Notice, how the channel was closed, but the items that were already send in it, were not lost, and the slow writer could still process them. |
Contributor
@spytheman Go permits to read from a closed channel, since permit to read from the closed channel the elements so far there with package main
import (
"fmt"
)
func main() {
data := make(chan int, 10)
end := make(chan bool)
go write(data, end)
<- end
go read(data, end)
<- end
}
func write(data chan int, end chan bool) {
for i := 0; i < 5; i++ {
data <- i
}
close(data)
fmt.Println("closed")
end <- true
}
func read(data chan int, end chan bool) {
fmt.Printf("Pending to read = %v\n", len(data))
for i := range data {
fmt.Printf("read %v\n", i)
}
end <- true
}I'm still trying to make a V equivalent... |
Contributor
|
In V also we can read from a closed channel. We call module main
fn main() {
data := chan int{ cap:10 }
end := chan bool{}
spawn write(data, end)
_ := <- end
spawn read(data, end)
_ := <- end
}
fn write(data chan int, end chan bool) {
for i := 0; i < 5; i++ {
data <- i
}
data.close()
println('writer: data closed')
end <- true
}
fn read(data chan int, end chan bool) {
for {
if select {
d := <- data {
println('read: d = ${d}')
}
} {
println('read: open')
} else {
println('read: closed')
break
}
}
end <- true
}Sounds weird but seems to work. |
Contributor
Author
|
I add a v run run_bench.vAnd this will generate a compile cmd is something like this: v channel_bench_v.v -cc clangSome results on my system(objs/µs, higher is better): before this PR:
after apply this PR:
My system:
|
Contributor
Author
|
remove gc, |
Contributor
|
Thank you. |
Contributor
|
Good work.
I tried to run both gcc resultsWith tcc, ... while the last line was |
Contributor
Author
|
On my system, all ok for |
Contributor
|
For me, the benchmark also shows an improvement in throughput for this PR: master_result.md
pr_24802_result.md
v doctor
|
Contributor
|
I should probably update my valgrind, or install a new Ubuntu one of these days 🤔 . |
Contributor
|
I want to test a bit more, but so far, it works very well. Excellent work. |
spytheman
approved these changes
Jun 28, 2025
Contributor
Author
|
I think following lines in mut expected_status := u16(BufferElemStat.unused)
for !C.atomic_compare_exchange_weak_u16(status_adr, &expected_status,
u16(BufferElemStat.writing)) {
expected_status = u16(BufferElemStat.unused)
}and mut expected_status := u16(BufferElemStat.written)
for !C.atomic_compare_exchange_weak_u16(status_adr, &expected_status,
u16(BufferElemStat.reading)) {
expected_status = u16(BufferElemStat.written)
}
Wait a moment, I will check this. |
Contributor
Author
|
After upgrade my $ cat ttt.v
module main
fn main() {
ch := chan int{cap: 10}
expected := 42
spawn fn [expected, ch] () {
ch <- expected
}()
received := <-ch
ch.close()
dump(received)
}
mars@mars-QiTianM428-A688:~/v/bug/channel$ cat run.sh
EXE=ttt
v -cc clang -gc none -cg -showcc -keepc -d valgrind ${EXE}.v
valgrind --tool=drd ./${EXE}
valgrind --tool=helgrind ./${EXE}
#v -cc clang -gc none -cg -showcc -keepc -cflags "-fsanitize=thread" ${EXE}.v
#./${EXE}
mars@mars-QiTianM428-A688:~/v/bug/channel$ ./run.sh
> C compiler cmd: 'clang' '@/tmp/v_1000/ttt.tmp.c.rsp'
> C compiler response file "/tmp/v_1000/ttt.tmp.c.rsp":
-fwrapv -g -O0 -o '/home/mars/v/bug/channel/ttt' -I "/media/HD/github/kbkpbot/v/thirdparty/stdatomic/nix" -I "/usr/include/valgrind" "/tmp/v_1000/ttt.tmp.c" -std=gnu99 -D_DEFAULT_SOURCE -rdynamic -lpthread
==722378== drd, a thread error detector
==722378== Copyright (C) 2006-2024, and GNU GPL'd, by Bart Van Assche et al.
==722378== Using Valgrind-3.25.1 and LibVEX; rerun with -h for copyright info
==722378== Command: ./ttt
==722378==
[ttt.v:15] received: 42
==722378==
==722378== For lists of detected and suppressed errors, rerun with: -s
==722378== ERROR SUMMARY: 0 errors from 0 contexts (suppressed: 14 from 10)
==722380== Helgrind, a thread error detector
==722380== Copyright (C) 2007-2024, and GNU GPL'd, by OpenWorks LLP et al.
==722380== Using Valgrind-3.25.1 and LibVEX; rerun with -h for copyright info
==722380== Command: ./ttt
==722380==
[ttt.v:15] received: 42
==722380==
==722380== Use --history-level=approx or =none to gain increased speed, at
==722380== the cost of reduced accuracy of conflicting-access information
==722380== For lists of detected and suppressed errors, rerun with: -s
==722380== ERROR SUMMARY: 0 errors from 0 contexts (suppressed: 0 from 0)
mars@mars-QiTianM428-A688:~/v/bug/channel$ valgrind --version
valgrind-3.25.1 |
Contributor
Author
No need to fix this any more. |
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
SpinLockfor channel, replace some atomic operations.valgrind, this will may debugging multi-thread app easier.BTW, I found that it can read from a closed channel, is this a correct behavior?