diff options
Diffstat (limited to 'src/mongo/util/concurrency/ticketholder.cpp')
| -rw-r--r-- | src/mongo/util/concurrency/ticketholder.cpp | 236 |
1 files changed, 119 insertions, 117 deletions
diff --git a/src/mongo/util/concurrency/ticketholder.cpp b/src/mongo/util/concurrency/ticketholder.cpp index ca0c2560ef0..17efed9300d 100644 --- a/src/mongo/util/concurrency/ticketholder.cpp +++ b/src/mongo/util/concurrency/ticketholder.cpp @@ -36,161 +36,163 @@ namespace mongo { #if defined(__linux__) - namespace { - void _check(int ret) { - if (ret == 0) - return; - int err = errno; - severe() << "error in Ticketholder: " << errnoWithDescription(err); - fassertFailed(28604); - } - } +namespace { +void _check(int ret) { + if (ret == 0) + return; + int err = errno; + severe() << "error in Ticketholder: " << errnoWithDescription(err); + fassertFailed(28604); +} +} - TicketHolder::TicketHolder(int num) - : _outof(num) { - _check(sem_init(&_sem, 0, num)); - } +TicketHolder::TicketHolder(int num) : _outof(num) { + _check(sem_init(&_sem, 0, num)); +} - TicketHolder::~TicketHolder(){ - _check(sem_destroy(&_sem)); - } +TicketHolder::~TicketHolder() { + _check(sem_destroy(&_sem)); +} - bool TicketHolder::tryAcquire() { - while (0 != sem_trywait(&_sem)) { - switch(errno) { - case EAGAIN: return false; - case EINTR: break; - default: _check(-1); - } +bool TicketHolder::tryAcquire() { + while (0 != sem_trywait(&_sem)) { + switch (errno) { + case EAGAIN: + return false; + case EINTR: + break; + default: + _check(-1); } - return true; } + return true; +} - void TicketHolder::waitForTicket() { - while (0 != sem_wait(&_sem)) { - switch(errno) { - case EINTR: break; - default: _check(-1); - } +void TicketHolder::waitForTicket() { + while (0 != sem_wait(&_sem)) { + switch (errno) { + case EINTR: + break; + default: + _check(-1); } } +} - void TicketHolder::release() { - _check(sem_post(&_sem)); - } - - Status TicketHolder::resize(int newSize) { - stdx::lock_guard<stdx::mutex> lk(_resizeMutex); +void TicketHolder::release() { + _check(sem_post(&_sem)); +} - if (newSize < 5) - return Status(ErrorCodes::BadValue, - str::stream() << "Minimum value for semaphore is 5; given " - << newSize); +Status TicketHolder::resize(int newSize) { + stdx::lock_guard<stdx::mutex> lk(_resizeMutex); - if (newSize > SEM_VALUE_MAX) - return Status(ErrorCodes::BadValue, - str::stream() << "Maximum value for semaphore is " - << SEM_VALUE_MAX << "; given " << newSize ); + if (newSize < 5) + return Status(ErrorCodes::BadValue, + str::stream() << "Minimum value for semaphore is 5; given " << newSize); - while (_outof.load() < newSize) { - release(); - _outof.fetchAndAdd(1); - } - - while (_outof.load() > newSize) { - waitForTicket(); - _outof.subtractAndFetch(1); - } + if (newSize > SEM_VALUE_MAX) + return Status(ErrorCodes::BadValue, + str::stream() << "Maximum value for semaphore is " << SEM_VALUE_MAX + << "; given " << newSize); - invariant(_outof.load() == newSize); - return Status::OK(); + while (_outof.load() < newSize) { + release(); + _outof.fetchAndAdd(1); } - int TicketHolder::available() const { - int val = 0; - _check(sem_getvalue(&_sem, &val)); - return val; + while (_outof.load() > newSize) { + waitForTicket(); + _outof.subtractAndFetch(1); } - int TicketHolder::used() const { - return outof() - available(); - } + invariant(_outof.load() == newSize); + return Status::OK(); +} - int TicketHolder::outof() const { - return _outof.load(); - } +int TicketHolder::available() const { + int val = 0; + _check(sem_getvalue(&_sem, &val)); + return val; +} + +int TicketHolder::used() const { + return outof() - available(); +} + +int TicketHolder::outof() const { + return _outof.load(); +} #else - TicketHolder::TicketHolder( int num ) : _outof(num), _num(num) {} +TicketHolder::TicketHolder(int num) : _outof(num), _num(num) {} - TicketHolder::~TicketHolder() = default; +TicketHolder::~TicketHolder() = default; - bool TicketHolder::tryAcquire() { - stdx::lock_guard<stdx::mutex> lk( _mutex ); - return _tryAcquire(); - } +bool TicketHolder::tryAcquire() { + stdx::lock_guard<stdx::mutex> lk(_mutex); + return _tryAcquire(); +} - void TicketHolder::waitForTicket() { - stdx::unique_lock<stdx::mutex> lk( _mutex ); +void TicketHolder::waitForTicket() { + stdx::unique_lock<stdx::mutex> lk(_mutex); - while( ! _tryAcquire() ) { - _newTicket.wait( lk ); - } + while (!_tryAcquire()) { + _newTicket.wait(lk); } +} - void TicketHolder::release() { - { - stdx::lock_guard<stdx::mutex> lk( _mutex ); - _num++; - } - _newTicket.notify_one(); +void TicketHolder::release() { + { + stdx::lock_guard<stdx::mutex> lk(_mutex); + _num++; } + _newTicket.notify_one(); +} - Status TicketHolder::resize( int newSize ) { - stdx::lock_guard<stdx::mutex> lk( _mutex ); +Status TicketHolder::resize(int newSize) { + stdx::lock_guard<stdx::mutex> lk(_mutex); - int used = _outof.load() - _num; - if ( used > newSize ) { - std::stringstream ss; - ss << "can't resize since we're using (" << used << ") " - << "more than newSize(" << newSize << ")"; + int used = _outof.load() - _num; + if (used > newSize) { + std::stringstream ss; + ss << "can't resize since we're using (" << used << ") " + << "more than newSize(" << newSize << ")"; - std::string errmsg = ss.str(); - log() << errmsg; - return Status(ErrorCodes::BadValue, errmsg); - } + std::string errmsg = ss.str(); + log() << errmsg; + return Status(ErrorCodes::BadValue, errmsg); + } - _outof.store(newSize); - _num = _outof.load() - used; + _outof.store(newSize); + _num = _outof.load() - used; - // Potentially wasteful, but easier to see is correct - _newTicket.notify_all(); - return Status::OK(); - } + // Potentially wasteful, but easier to see is correct + _newTicket.notify_all(); + return Status::OK(); +} - int TicketHolder::available() const { - return _num; - } +int TicketHolder::available() const { + return _num; +} - int TicketHolder::used() const { - return outof() - _num; - } +int TicketHolder::used() const { + return outof() - _num; +} - int TicketHolder::outof() const { - return _outof.load(); - } +int TicketHolder::outof() const { + return _outof.load(); +} - bool TicketHolder::_tryAcquire(){ - if ( _num <= 0 ) { - if ( _num < 0 ) { - std::cerr << "DISASTER! in TicketHolder" << std::endl; - } - return false; +bool TicketHolder::_tryAcquire() { + if (_num <= 0) { + if (_num < 0) { + std::cerr << "DISASTER! in TicketHolder" << std::endl; } - _num--; - return true; + return false; } + _num--; + return true; +} #endif - } |