ed/Odin
1
0
Fork 0
mirror of https://github.com/Ed94/Odin.git synced 2026-08-06 15:48:51 +00:00
Code Issues Packages Projects Releases Wiki Activity

Improve sync.Channel behaviour

Browse Source
This commit is contained in:
gingerBill
2020-07-14 18:43:50 +01:00
parent 86f1574f78
commit 96ad6d2084
1 changed files with 28 additions and 18 deletions
Download Patch File Download Diff File Expand all files Collapse all files
core/sync/channel.odin
+28 -18
View File
@@ -1,5 +1,6 @@
package sync package sync
// import "core:fmt"
import "core:mem" import "core:mem"
import "core:time" import "core:time"
import "core:intrinsics" import "core:intrinsics"
@@ -39,11 +40,11 @@ channel_cap :: proc(ch: $C/Channel($T)) -> int {
channel_send :: proc(ch: $C/Channel($T), msg: T, loc := #caller_location) { channel_send :: proc(ch: $C/Channel($T), msg: T, loc := #caller_location) {
msg := msg; msg := msg;
_ = raw_channel_send_impl(ch._internal, &msg, false, loc); _ = raw_channel_send_impl(ch._internal, &msg, /*block*/true, loc);
} }
channel_try_send :: proc(ch: $C/Channel($T), msg: T, loc := #caller_location) -> bool { channel_try_send :: proc(ch: $C/Channel($T), msg: T, loc := #caller_location) -> bool {
msg := msg; msg := msg;
return raw_channel_send_impl(ch._internal, &msg, true, loc); return raw_channel_send_impl(ch._internal, &msg, /*block*/false, loc);
} }
channel_recv :: proc(ch: $C/Channel($T), loc := #caller_location) -> (msg: T) { channel_recv :: proc(ch: $C/Channel($T), loc := #caller_location) -> (msg: T) {
@@ -68,6 +69,10 @@ channel_try_recv :: proc(ch: $C/Channel($T), loc := #caller_location) -> (msg: T
channel_is_nil :: proc(ch: $C/Channel($T)) -> bool { channel_is_nil :: proc(ch: $C/Channel($T)) -> bool {
return ch._internal == nil; return ch._internal == nil;
} }
channel_is_open :: proc(ch: $C/Channel($T)) -> bool {
c := ch._internal;
return c != nil && !c.closed;
}
channel_eq :: proc(a, b: $C/Channel($T)) -> bool { channel_eq :: proc(a, b: $C/Channel($T)) -> bool {
@@ -108,19 +113,14 @@ channel_close :: proc(ch: $C/Channel($T), loc := #caller_location) {
} }
channel_iterator :: proc(ch: $C/Channel($T)) -> (val: T, open: bool) { channel_iterator :: proc(ch: $C/Channel($T)) -> (val: T, ok: bool) {
c := ch._internal; c := ch._internal;
switch { if c == nil {
case c == nil:
return; return;
case intrinsics.atomic_load(&c.closed): }
if channel_can_recv(ch) {
val = channel_recv(ch); if !c.closed || c.len > 0 {
open = true; val, ok = channel_recv(ch), true;
}
case:
val = channel_recv(ch);
open = true;
} }
return; return;
} }
@@ -129,8 +129,11 @@ channel_iterator :: proc(ch: $C/Channel($T)) -> (val: T, open: bool) {
channel_select_recv :: proc(channels: ..^Raw_Channel) -> (index: int) { channel_select_recv :: proc(channels: ..^Raw_Channel) -> (index: int) {
backing: [64]int; backing: [64]int;
candidates := backing[:]; candidates := backing[:];
if len(channels) > len(backing) { cap := len(channels);
candidates = make([]int, len(channels), context.temp_allocator); if cap > len(backing) {
candidates = make([]int, cap, context.temp_allocator);
} else {
candidates = candidates[:cap];
} }
count := u32(0); count := u32(0);
@@ -298,7 +301,7 @@ raw_channel_destroy :: proc(c: ^Raw_Channel) {
} }
raw_channel_send_impl :: proc(c: ^Raw_Channel, msg: rawptr, no_block: bool, loc := #caller_location) -> bool { raw_channel_send_impl :: proc(c: ^Raw_Channel, msg: rawptr, block: bool, loc := #caller_location) -> bool {
send :: proc(c: ^Raw_Channel, src: rawptr) { send :: proc(c: ^Raw_Channel, src: rawptr) {
dst := uintptr(c.data) + uintptr(c.write * c.elem_size); dst := uintptr(c.data) + uintptr(c.write * c.elem_size);
mem.copy(rawptr(dst), src, c.elem_size); mem.copy(rawptr(dst), src, c.elem_size);
@@ -315,7 +318,7 @@ raw_channel_send_impl :: proc(c: ^Raw_Channel, msg: rawptr, no_block: bool, loc
mutex_lock(&c.mutex); mutex_lock(&c.mutex);
if c.cap > 0 { if c.cap > 0 {
if no_block && c.len >= c.cap { if !block && c.len >= c.cap {
mutex_unlock(&c.mutex); mutex_unlock(&c.mutex);
return false; return false;
} }
@@ -323,6 +326,8 @@ raw_channel_send_impl :: proc(c: ^Raw_Channel, msg: rawptr, no_block: bool, loc
for c.len >= c.cap { for c.len >= c.cap {
condition_wait_for(&c.cond); condition_wait_for(&c.cond);
} }
} else if c.len > 0 {
condition_wait_for(&c.cond);
} }
send(c, msg); send(c, msg);
@@ -352,7 +357,12 @@ raw_channel_recv_impl :: proc(c: ^Raw_Channel, res: rawptr, loc := #caller_locat
} }
intrinsics.atomic_store(&c.ready, false); intrinsics.atomic_store(&c.ready, false);
recv(c, res, loc); recv(c, res, loc);
if c.cap > 0 && c.len == c.cap - 1 { if c.cap > 0 {
if c.len == c.cap - 1 {
// NOTE(bill): Only signal on the last one
condition_signal(&c.cond);
}
} else {
condition_signal(&c.cond); condition_signal(&c.cond);
} }
} }