diff --git a/thirdparty/stdatomic/win/atomic.h b/thirdparty/stdatomic/win/atomic.h index 698a51c495d9f2..18879d21c90bd2 100644 --- a/thirdparty/stdatomic/win/atomic.h +++ b/thirdparty/stdatomic/win/atomic.h @@ -123,7 +123,7 @@ __CRT_INLINE SHORT _InterlockedExchangeAdd16(SHORT volatile *Addend, SHORT Value #define InterlockedIncrement64 _InterlockedExchangeAdd64 __CRT_INLINE VOID __faststorefence() { - __asm__ __volatile__ ("sfence"); + __asm__ __volatile__ ("mfence"); } #endif diff --git a/vlib/v/tests/bench/channel_bench_go.go b/vlib/sync/bench/channel_bench_go.go similarity index 100% rename from vlib/v/tests/bench/channel_bench_go.go rename to vlib/sync/bench/channel_bench_go.go diff --git a/vlib/v/tests/bench/channel_bench_v.v b/vlib/sync/bench/channel_bench_v.v similarity index 100% rename from vlib/v/tests/bench/channel_bench_v.v rename to vlib/sync/bench/channel_bench_v.v diff --git a/vlib/v/tests/channel_1_test.v b/vlib/sync/channel_1_test.v similarity index 100% rename from vlib/v/tests/channel_1_test.v rename to vlib/sync/channel_1_test.v diff --git a/vlib/v/tests/channel_2_test.v b/vlib/sync/channel_2_test.v similarity index 100% rename from vlib/v/tests/channel_2_test.v rename to vlib/sync/channel_2_test.v diff --git a/vlib/v/tests/channel_3_test.v b/vlib/sync/channel_3_test.v similarity index 100% rename from vlib/v/tests/channel_3_test.v rename to vlib/sync/channel_3_test.v diff --git a/vlib/v/tests/channel_4_test.v b/vlib/sync/channel_4_test.v similarity index 100% rename from vlib/v/tests/channel_4_test.v rename to vlib/sync/channel_4_test.v diff --git a/vlib/v/tests/channel_select_test.v b/vlib/sync/channel_select_test.v similarity index 96% rename from vlib/v/tests/channel_select_test.v rename to vlib/sync/channel_select_test.v index 05db0cbb018acf..10e2a0ec8ce3c6 100644 --- a/vlib/v/tests/channel_select_test.v +++ b/vlib/sync/channel_select_test.v @@ -40,7 +40,7 @@ fn test_select() { go do_send_byte(mut chb) go do_send_i64(mut chl) mut channels := [chi, recch, chl, chb] - directions := [false, true, false, false] + directions := [sync.Direction.pop, .push, .pop, .pop] mut sum := i64(0) mut rl := i64(0) mut ri := int(0) diff --git a/vlib/sync/channels.v b/vlib/sync/channels.v index dd7953a5bee525..6f0f76d21fea90 100644 --- a/vlib/sync/channels.v +++ b/vlib/sync/channels.v @@ -71,6 +71,11 @@ mut: nxt &Subscription } +enum Direction { + pop + push +} + struct Channel { writesem Semaphore // to wake thread that wanted to write, but buffer was full readsem Semaphore // to wake thread that wanted to read, but buffer was empty @@ -97,17 +102,18 @@ mut: // atomic } pub fn new_channel(n u32) &Channel { + st := sizeof(T) return &Channel{ writesem: new_semaphore_init(if n > 0 { n + 1 } else { 1 }) readsem: new_semaphore_init(if n > 0 { u32(0) } else { 1 }) writesem_im: new_semaphore() readsem_im: new_semaphore() - objsize: sizeof(T) + objsize: st queue_length: n write_free: n read_avail: 0 - ringbuf: if n > 0 { malloc(int(n * sizeof(T))) } else { byteptr(0) } - statusbuf: if n > 0 { vcalloc(int(n * sizeof(u16))) } else { byteptr(0) } + ringbuf: if n > 0 { malloc(int(n * st)) } else { byteptr(0) } + statusbuf: if n > 0 { vcalloc(int(n * 2)) } else { byteptr(0) } write_subscriber: 0 read_subscriber: 0 } @@ -412,13 +418,13 @@ fn (mut ch Channel) try_pop(dest voidptr, no_block bool) bool { // Wait `timeout` on any of `channels[i]` until one of them can push (`is_push[i] = true`) or pop (`is_push[i] = false`) // object referenced by `objrefs[i]`. `timeout = 0` means wait unlimited time -pub fn channel_select(mut channels []&Channel, is_push []bool, mut objrefs []voidptr, timeout time.Duration) int { - assert channels.len == is_push.len - assert is_push.len == objrefs.len +pub fn channel_select(mut channels []&Channel, dir []Direction, mut objrefs []voidptr, timeout time.Duration) int { + assert channels.len == dir.len + assert dir.len == objrefs.len mut subscr := []Subscription{len: channels.len} sem := new_semaphore() for i, ch in channels { - if is_push[i] { + if dir[i] == .push { mut null16 := u16(0) for !C.atomic_compare_exchange_weak_u16(&ch.write_sub_mtx, &null16, u16(1)) { null16 = u16(0) @@ -453,7 +459,7 @@ pub fn channel_select(mut channels []&Channel, is_push []bool, mut objrefs []voi if i >= channels.len { i -= channels.len } - if is_push[i] { + if dir[i] == .push { if channels[i].try_push(objrefs[i], true) { event_idx = i goto restore @@ -477,7 +483,7 @@ pub fn channel_select(mut channels []&Channel, is_push []bool, mut objrefs []voi restore: // reset subscribers for i, ch in channels { - if is_push[i] { + if dir[i] == .push { mut null16 := u16(0) for !C.atomic_compare_exchange_weak_u16(&ch.write_sub_mtx, &null16, u16(1)) { null16 = u16(0)