hirun 0.1.8

A concurrent framework for asynchronous programming based on event-driven, non-blocking I/O mechanism
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136

use super::Attr;
use core::sync::atomic::{
    fence, AtomicU8,
    Ordering::{self, Acquire, Relaxed, Release},
};

#[repr(C)]
pub(crate) struct AtomicStatus {
    stat: AtomicU8,
    rcnt: AtomicU8,
    wcnt: AtomicU8,
    wexp: u8,
    group: u8,
    sext: u8,
    worker: u16,
}

pub(crate) const STAT_RUN: u8 = 0;
pub(crate) const STAT_JOIN: u8 = 1;
pub(crate) const STAT_RETURN: u8 = 2;
pub(crate) const STAT_FINISH: u8 = 3;
pub(crate) const STAT_ABORT: u8 = 4;
pub(crate) const STAT_TIMEOUT: u8 = 5;

const PRI_SHIFT: u8 = 0;
const YIELD_SHIFT: u8 = 1;

impl AtomicStatus {
    pub(crate) fn new(attr: &Attr) -> Self {
        Self {
            stat: AtomicU8::new(STAT_RUN),
            rcnt: AtomicU8::new(1),
            wcnt: AtomicU8::new(1),
            wexp: 1,
            group: attr.group_id,
            sext: sext_new(0, PRI_SHIFT, attr.priority != 0),
            worker: 0,
        }
    }

    pub(crate) fn test_dec_wake(&self, cnt: u8) -> bool {
        let old = self.wcnt.fetch_sub(cnt, Release);
        (old - cnt) >= self.wexp
    }

    pub(crate) fn inc_wake(&self) {
        self.wcnt.fetch_add(1, Relaxed);
    }

    pub(crate) fn test_inc_wake(&self) -> bool {
        let cnt = self.wcnt.fetch_add(1, Acquire);
        cnt == self.wexp - 1
    }

    pub(crate) fn wake_expect(&self) -> u8 {
        assert!(self.wcnt.load(Relaxed) >= self.wexp);
        self.wexp
    }

    pub(crate) fn set_wake_expect(&mut self, wexp: u8) {
        assert!(wexp > 0);
        self.wexp = wexp;
    }

    pub(crate) fn inc_ref(&self) {
        let cnt = self.rcnt.fetch_add(1, Relaxed);
        assert!(cnt < 254);
        assert!(cnt > 0);
    }

    pub(crate) fn test_dec_ref(&self) -> bool {
        if self.rcnt.fetch_sub(1, Release) == 1 {
            fence(Acquire);
            return true;
        }
        false
    }

    pub(crate) fn status(&self, order: Ordering) -> u8 {
        self.stat.load(order)
    }

    pub(crate) fn set_status(&self, new: u8, order: Ordering) {
        self.stat.store(new, order);
    }

    pub(crate) fn cmp_xchg_status(
        &self,
        current: u8,
        new: u8,
        set_order: Ordering,
    ) -> Result<u8, u8> {
        self.stat.compare_exchange(current, new, set_order, Relaxed)
    }

    pub(crate) fn priority(&self) -> u8 {
        if sext_get(self.sext, PRI_SHIFT) {
            1
        } else {
            0
        }
    }
    pub(crate) fn group(&self) -> u8 {
        self.group
    }

    pub(crate) fn set_yield(&mut self) {
        self.sext = sext_new(self.sext, YIELD_SHIFT, true);
    }

    pub(crate) fn is_yield(&self) -> bool {
        sext_get(self.sext, YIELD_SHIFT)
    }

    pub(crate) fn get_local(&self) -> Option<u16> {
        if (self.worker & 0x8000) > 0 {
            return Some(self.worker & !0x8000);
        }
        None
    }
    pub(crate) fn set_local(&mut self, id: u16) {
        self.worker = id | 0x8000;
    }
}

fn sext_new(old: u8, shift: u8, val: bool) -> u8 {
    if val {
        old | (0x01 << shift)
    } else {
        old & !(0x01 << shift)
    }
}

fn sext_get(sext: u8, shift: u8) -> bool {
    (sext & (0x01 << shift)) != 0
}