mirror/v
1
0
Fork 0
mirror of https://github.com/vlang/v.git synced 2023-08-10 21:13:21 +03:00
Code Issues Projects Releases Wiki Activity

channels: refactor the channel_select function (#18711)

Browse Source
This commit is contained in:
Herman 2023-07-02 15:45:30 +03:00 committed by GitHub
parent 329e063752
commit af38f8b3ba
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
1 changed files with 40 additions and 68 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

108
vlib/sync/channels.c.v
View File

@ -556,35 +556,24 @@ pub fn channel_select(mut channels []&Channel, dir []Direction, mut objrefs []vo
sem.init(0) sem.init(0)
for i, ch in channels { for i, ch in channels {
subscr[i].sem = unsafe { &sem } subscr[i].sem = unsafe { &sem }
if dir[i] == .push { sub_mtx, subscriber := if dir[i] == .push {
mut null16 := u16(0) &ch.write_sub_mtx, &ch.write_subscriber
for !C.atomic_compare_exchange_weak_u16(&ch.write_sub_mtx, &null16, u16(1)) {
null16 = u16(0)
}
subscr[i].prev = unsafe { &ch.write_subscriber }
unsafe {
subscr[i].nxt = &Subscription(C.atomic_exchange_ptr(&voidptr(&ch.write_subscriber),
&subscr[i]))
}
if voidptr(subscr[i].nxt) != unsafe { nil } {
subscr[i].nxt.prev = unsafe { &subscr[i].nxt }
}
C.atomic_store_u16(&ch.write_sub_mtx, u16(0))
} else { } else {
mut null16 := u16(0) &ch.read_sub_mtx, &ch.read_subscriber
for !C.atomic_compare_exchange_weak_u16(&ch.read_sub_mtx, &null16, u16(1)) {
null16 = u16(0)
}
subscr[i].prev = unsafe { &ch.read_subscriber }
unsafe {
subscr[i].nxt = &Subscription(C.atomic_exchange_ptr(&voidptr(&ch.read_subscriber),
&subscr[i]))
}
if voidptr(subscr[i].nxt) != unsafe { nil } {
subscr[i].nxt.prev = unsafe { &subscr[i].nxt }
}
C.atomic_store_u16(&ch.read_sub_mtx, u16(0))
} }
mut null16 := u16(0)
for !C.atomic_compare_exchange_weak_u16(sub_mtx, &null16, u16(1)) {
null16 = u16(0)
}
subscr[i].prev = unsafe { subscriber }
unsafe {
subscr[i].nxt = &Subscription(C.atomic_exchange_ptr(&voidptr(subscriber),
&subscr[i]))
}
if voidptr(subscr[i].nxt) != unsafe { nil } {
subscr[i].nxt.prev = unsafe { &subscr[i].nxt }
}
C.atomic_store_u16(sub_mtx, u16(0))
} }
stopwatch := if timeout == time.infinite || timeout <= 0 { stopwatch := if timeout == time.infinite || timeout <= 0 {
time.StopWatch{} time.StopWatch{}
@ -601,22 +590,16 @@ pub fn channel_select(mut channels []&Channel, dir []Direction, mut objrefs []vo
if i >= channels.len { if i >= channels.len {
i -= channels.len i -= channels.len
} }
if dir[i] == .push { stat := if dir[i] == .push {
stat := channels[i].try_push_priv(objrefs[i], true) channels[i].try_push_priv(objrefs[i], true)
if stat == .success {
event_idx = i
break outer
} else if stat == .closed {
num_closed++
}
} else { } else {
stat := channels[i].try_pop_priv(objrefs[i], true) channels[i].try_pop_priv(objrefs[i], true)
if stat == .success { }
event_idx = i if stat == .success {
break outer event_idx = i
} else if stat == .closed { break outer
num_closed++ } else if stat == .closed {
} num_closed++
} }
} }
if num_closed == channels.len { if num_closed == channels.len {
@ -637,34 +620,23 @@ pub fn channel_select(mut channels []&Channel, dir []Direction, mut objrefs []vo
} }
// reset subscribers // reset subscribers
for i, ch in channels { for i, ch in channels {
if dir[i] == .push { sub_mtx := if dir[i] == .push {
mut null16 := u16(0) &ch.write_sub_mtx
for !C.atomic_compare_exchange_weak_u16(&ch.write_sub_mtx, &null16, u16(1)) {
null16 = u16(0)
}
unsafe {
*subscr[i].prev = subscr[i].nxt
}
if unsafe { subscr[i].nxt != 0 } {
subscr[i].nxt.prev = subscr[i].prev
// just in case we have missed a semaphore during restore
subscr[i].nxt.sem.post()
}
C.atomic_store_u16(&ch.write_sub_mtx, u16(0))
} else { } else {
mut null16 := u16(0) &ch.read_sub_mtx
for !C.atomic_compare_exchange_weak_u16(&ch.read_sub_mtx, &null16, u16(1)) {
null16 = u16(0)
}
unsafe {
*subscr[i].prev = subscr[i].nxt
}
if unsafe { subscr[i].nxt != 0 } {
subscr[i].nxt.prev = subscr[i].prev
subscr[i].nxt.sem.post()
}
C.atomic_store_u16(&ch.read_sub_mtx, u16(0))
} }
mut null16 := u16(0)
for !C.atomic_compare_exchange_weak_u16(sub_mtx, &null16, u16(1)) {
null16 = u16(0)
}
unsafe {
*subscr[i].prev = subscr[i].nxt
}
if unsafe { subscr[i].nxt != 0 } {
subscr[i].nxt.prev = subscr[i].prev
subscr[i].nxt.sem.post()
}
C.atomic_store_u16(sub_mtx, u16(0))
} }
sem.destroy() sem.destroy()
return event_idx return event_idx