countdown_event.cpp 3.21 KB
Newer Older
gejun's avatar
gejun committed
1
// bthread - A M:N threading library to make applications more concurrent.
gejun's avatar
gejun committed
2
// Copyright (c) 2016 Baidu, Inc.
gejun's avatar
gejun committed
3 4 5 6 7 8 9 10 11 12 13 14
// 
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
// 
//     http://www.apache.org/licenses/LICENSE-2.0
// 
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
gejun's avatar
gejun committed
15 16 17 18

// Author: Zhangyi Chen (chenzhangyi01@baidu.com)
// Date: 2016/06/03 13:15:24

19
#include "butil/atomicops.h"     // butil::atomic<int>
gejun's avatar
gejun committed
20
#include "bthread/butex.h"
gejun's avatar
gejun committed
21 22 23 24
#include "bthread/countdown_event.h"

namespace bthread {

gejun's avatar
gejun committed
25 26 27 28 29 30 31
CountdownEvent::CountdownEvent(int initial_count) {
    if (initial_count < 0) {
        LOG(FATAL) << "Invalid initial_count=" << initial_count;
        abort();
    }
    _butex = butex_create_checked<int>();
    *_butex = initial_count;
gejun's avatar
gejun committed
32 33 34 35
    _wait_was_invoked = false;
}

CountdownEvent::~CountdownEvent() {
gejun's avatar
gejun committed
36
    butex_destroy(_butex);
gejun's avatar
gejun committed
37 38 39
}

void CountdownEvent::signal(int sig) {
40 41 42
    // Have to save _butex, *this is probably defreferenced by the wait thread
    // which sees fetch_sub
    void* const saved_butex = _butex;
43 44
    const int prev = ((butil::atomic<int>*)_butex)
        ->fetch_sub(sig, butil::memory_order_release);
45
    // DON'T touch *this ever after
gejun's avatar
gejun committed
46 47 48 49
    if (prev > sig) {
        return;
    }
    LOG_IF(ERROR, prev < sig) << "Counter is over decreased";
50
    butex_wake_all(saved_butex);
gejun's avatar
gejun committed
51 52
}

53
int CountdownEvent::wait() {
gejun's avatar
gejun committed
54 55 56
    _wait_was_invoked = true;
    for (;;) {
        const int seen_counter = 
57
            ((butil::atomic<int>*)_butex)->load(butil::memory_order_acquire);
gejun's avatar
gejun committed
58
        if (seen_counter <= 0) {
59 60 61 62 63
            return 0;
        }
        if (butex_wait(_butex, seen_counter, NULL) < 0 &&
            errno != EWOULDBLOCK && errno != EINTR) {
            return errno;
gejun's avatar
gejun committed
64 65 66 67 68 69
        }
    }
}

void CountdownEvent::add_count(int v) {
    if (v <= 0) {
gejun's avatar
gejun committed
70
        LOG_IF(ERROR, v < 0) << "Invalid count=" << v;
gejun's avatar
gejun committed
71 72 73 74
        return;
    }
    LOG_IF(ERROR, _wait_was_invoked) 
            << "Invoking add_count() after wait() was invoked";
75
    ((butil::atomic<int>*)_butex)->fetch_add(v, butil::memory_order_release);
gejun's avatar
gejun committed
76 77 78
}

void CountdownEvent::reset(int v) {
gejun's avatar
gejun committed
79 80 81 82
    if (v < 0) {
        LOG(ERROR) << "Invalid count=" << v;
        return;
    }
gejun's avatar
gejun committed
83
    const int prev_counter =
84 85
            ((butil::atomic<int>*)_butex)
                ->exchange(v, butil::memory_order_release);
gejun's avatar
gejun committed
86 87
    LOG_IF(ERROR, _wait_was_invoked && prev_counter)
        << "Invoking reset() while count=" << prev_counter;
gejun's avatar
gejun committed
88 89 90 91 92 93 94
    _wait_was_invoked = false;
}

int CountdownEvent::timed_wait(const timespec& duetime) {
    _wait_was_invoked = true;
    for (;;) {
        const int seen_counter = 
95
            ((butil::atomic<int>*)_butex)->load(butil::memory_order_acquire);
gejun's avatar
gejun committed
96 97 98
        if (seen_counter <= 0) {
            return 0;
        }
99 100
        if (butex_wait(_butex, seen_counter, &duetime) < 0 &&
            errno != EWOULDBLOCK && errno != EINTR) {
gejun's avatar
gejun committed
101
            return errno;
gejun's avatar
gejun committed
102 103 104 105 106
        }
    }
}

}  // namespace bthread