mirror of https://github.com/grpc/grpc.git
Merge pull request #20119 from markdroth/c++_mpscq
Convert mpscq API from C to C++.pull/15577/head^2
Mark D. Roth
5 years ago
committed by
GitHub
commit
d1bca6ff92
31 changed files with 438 additions and 410 deletions
@ -1,117 +0,0 @@ |
|||||||
/*
|
|
||||||
* |
|
||||||
* Copyright 2016 gRPC authors. |
|
||||||
* |
|
||||||
* 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. |
|
||||||
* |
|
||||||
*/ |
|
||||||
|
|
||||||
#include <grpc/support/port_platform.h> |
|
||||||
|
|
||||||
#include "src/core/lib/gpr/mpscq.h" |
|
||||||
|
|
||||||
#include <grpc/support/log.h> |
|
||||||
|
|
||||||
void gpr_mpscq_init(gpr_mpscq* q) { |
|
||||||
gpr_atm_no_barrier_store(&q->head, (gpr_atm)&q->stub); |
|
||||||
q->tail = &q->stub; |
|
||||||
gpr_atm_no_barrier_store(&q->stub.next, (gpr_atm)NULL); |
|
||||||
} |
|
||||||
|
|
||||||
void gpr_mpscq_destroy(gpr_mpscq* q) { |
|
||||||
GPR_ASSERT(gpr_atm_no_barrier_load(&q->head) == (gpr_atm)&q->stub); |
|
||||||
GPR_ASSERT(q->tail == &q->stub); |
|
||||||
} |
|
||||||
|
|
||||||
bool gpr_mpscq_push(gpr_mpscq* q, gpr_mpscq_node* n) { |
|
||||||
gpr_atm_no_barrier_store(&n->next, (gpr_atm)NULL); |
|
||||||
gpr_mpscq_node* prev = |
|
||||||
(gpr_mpscq_node*)gpr_atm_full_xchg(&q->head, (gpr_atm)n); |
|
||||||
gpr_atm_rel_store(&prev->next, (gpr_atm)n); |
|
||||||
return prev == &q->stub; |
|
||||||
} |
|
||||||
|
|
||||||
gpr_mpscq_node* gpr_mpscq_pop(gpr_mpscq* q) { |
|
||||||
bool empty; |
|
||||||
return gpr_mpscq_pop_and_check_end(q, &empty); |
|
||||||
} |
|
||||||
|
|
||||||
gpr_mpscq_node* gpr_mpscq_pop_and_check_end(gpr_mpscq* q, bool* empty) { |
|
||||||
gpr_mpscq_node* tail = q->tail; |
|
||||||
gpr_mpscq_node* next = (gpr_mpscq_node*)gpr_atm_acq_load(&tail->next); |
|
||||||
if (tail == &q->stub) { |
|
||||||
// indicates the list is actually (ephemerally) empty
|
|
||||||
if (next == nullptr) { |
|
||||||
*empty = true; |
|
||||||
return nullptr; |
|
||||||
} |
|
||||||
q->tail = next; |
|
||||||
tail = next; |
|
||||||
next = (gpr_mpscq_node*)gpr_atm_acq_load(&tail->next); |
|
||||||
} |
|
||||||
if (next != nullptr) { |
|
||||||
*empty = false; |
|
||||||
q->tail = next; |
|
||||||
return tail; |
|
||||||
} |
|
||||||
gpr_mpscq_node* head = (gpr_mpscq_node*)gpr_atm_acq_load(&q->head); |
|
||||||
if (tail != head) { |
|
||||||
*empty = false; |
|
||||||
// indicates a retry is in order: we're still adding
|
|
||||||
return nullptr; |
|
||||||
} |
|
||||||
gpr_mpscq_push(q, &q->stub); |
|
||||||
next = (gpr_mpscq_node*)gpr_atm_acq_load(&tail->next); |
|
||||||
if (next != nullptr) { |
|
||||||
*empty = false; |
|
||||||
q->tail = next; |
|
||||||
return tail; |
|
||||||
} |
|
||||||
// indicates a retry is in order: we're still adding
|
|
||||||
*empty = false; |
|
||||||
return nullptr; |
|
||||||
} |
|
||||||
|
|
||||||
void gpr_locked_mpscq_init(gpr_locked_mpscq* q) { |
|
||||||
gpr_mpscq_init(&q->queue); |
|
||||||
gpr_mu_init(&q->mu); |
|
||||||
} |
|
||||||
|
|
||||||
void gpr_locked_mpscq_destroy(gpr_locked_mpscq* q) { |
|
||||||
gpr_mpscq_destroy(&q->queue); |
|
||||||
gpr_mu_destroy(&q->mu); |
|
||||||
} |
|
||||||
|
|
||||||
bool gpr_locked_mpscq_push(gpr_locked_mpscq* q, gpr_mpscq_node* n) { |
|
||||||
return gpr_mpscq_push(&q->queue, n); |
|
||||||
} |
|
||||||
|
|
||||||
gpr_mpscq_node* gpr_locked_mpscq_try_pop(gpr_locked_mpscq* q) { |
|
||||||
if (gpr_mu_trylock(&q->mu)) { |
|
||||||
gpr_mpscq_node* n = gpr_mpscq_pop(&q->queue); |
|
||||||
gpr_mu_unlock(&q->mu); |
|
||||||
return n; |
|
||||||
} |
|
||||||
return nullptr; |
|
||||||
} |
|
||||||
|
|
||||||
gpr_mpscq_node* gpr_locked_mpscq_pop(gpr_locked_mpscq* q) { |
|
||||||
gpr_mu_lock(&q->mu); |
|
||||||
bool empty = false; |
|
||||||
gpr_mpscq_node* n; |
|
||||||
do { |
|
||||||
n = gpr_mpscq_pop_and_check_end(&q->queue, &empty); |
|
||||||
} while (n == nullptr && !empty); |
|
||||||
gpr_mu_unlock(&q->mu); |
|
||||||
return n; |
|
||||||
} |
|
||||||
@ -1,88 +0,0 @@ |
|||||||
/*
|
|
||||||
* |
|
||||||
* Copyright 2016 gRPC authors. |
|
||||||
* |
|
||||||
* 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. |
|
||||||
* |
|
||||||
*/ |
|
||||||
|
|
||||||
#ifndef GRPC_CORE_LIB_GPR_MPSCQ_H |
|
||||||
#define GRPC_CORE_LIB_GPR_MPSCQ_H |
|
||||||
|
|
||||||
#include <grpc/support/port_platform.h> |
|
||||||
|
|
||||||
#include <grpc/support/atm.h> |
|
||||||
#include <grpc/support/sync.h> |
|
||||||
#include <stdbool.h> |
|
||||||
#include <stddef.h> |
|
||||||
|
|
||||||
// Multiple-producer single-consumer lock free queue, based upon the
|
|
||||||
// implementation from Dmitry Vyukov here:
|
|
||||||
// http://www.1024cores.net/home/lock-free-algorithms/queues/intrusive-mpsc-node-based-queue
|
|
||||||
|
|
||||||
// List node (include this in a data structure at the top, and add application
|
|
||||||
// fields after it - to simulate inheritance)
|
|
||||||
typedef struct gpr_mpscq_node { |
|
||||||
gpr_atm next; |
|
||||||
} gpr_mpscq_node; |
|
||||||
|
|
||||||
// Actual queue type
|
|
||||||
typedef struct gpr_mpscq { |
|
||||||
// make sure head & tail don't share a cacheline
|
|
||||||
union { |
|
||||||
char padding[GPR_CACHELINE_SIZE]; |
|
||||||
gpr_atm head; |
|
||||||
}; |
|
||||||
gpr_mpscq_node* tail; |
|
||||||
gpr_mpscq_node stub; |
|
||||||
} gpr_mpscq; |
|
||||||
|
|
||||||
void gpr_mpscq_init(gpr_mpscq* q); |
|
||||||
void gpr_mpscq_destroy(gpr_mpscq* q); |
|
||||||
// Push a node
|
|
||||||
// Thread safe - can be called from multiple threads concurrently
|
|
||||||
// Returns true if this was possibly the first node (may return true
|
|
||||||
// sporadically, will not return false sporadically)
|
|
||||||
bool gpr_mpscq_push(gpr_mpscq* q, gpr_mpscq_node* n); |
|
||||||
// Pop a node (returns NULL if no node is ready - which doesn't indicate that
|
|
||||||
// the queue is empty!!)
|
|
||||||
// Thread compatible - can only be called from one thread at a time
|
|
||||||
gpr_mpscq_node* gpr_mpscq_pop(gpr_mpscq* q); |
|
||||||
// Pop a node; sets *empty to true if the queue is empty, or false if it is not
|
|
||||||
gpr_mpscq_node* gpr_mpscq_pop_and_check_end(gpr_mpscq* q, bool* empty); |
|
||||||
|
|
||||||
// An mpscq with a lock: it's safe to pop from multiple threads, but doing
|
|
||||||
// only one thread will succeed concurrently
|
|
||||||
typedef struct gpr_locked_mpscq { |
|
||||||
gpr_mpscq queue; |
|
||||||
gpr_mu mu; |
|
||||||
} gpr_locked_mpscq; |
|
||||||
|
|
||||||
void gpr_locked_mpscq_init(gpr_locked_mpscq* q); |
|
||||||
void gpr_locked_mpscq_destroy(gpr_locked_mpscq* q); |
|
||||||
// Push a node
|
|
||||||
// Thread safe - can be called from multiple threads concurrently
|
|
||||||
// Returns true if this was possibly the first node (may return true
|
|
||||||
// sporadically, will not return false sporadically)
|
|
||||||
bool gpr_locked_mpscq_push(gpr_locked_mpscq* q, gpr_mpscq_node* n); |
|
||||||
|
|
||||||
// Pop a node (returns NULL if no node is ready - which doesn't indicate that
|
|
||||||
// the queue is empty!!)
|
|
||||||
// Thread safe - can be called from multiple threads concurrently
|
|
||||||
gpr_mpscq_node* gpr_locked_mpscq_try_pop(gpr_locked_mpscq* q); |
|
||||||
|
|
||||||
// Pop a node. Returns NULL only if the queue was empty at some point after
|
|
||||||
// calling this function
|
|
||||||
gpr_mpscq_node* gpr_locked_mpscq_pop(gpr_locked_mpscq* q); |
|
||||||
|
|
||||||
#endif /* GRPC_CORE_LIB_GPR_MPSCQ_H */ |
|
||||||
@ -0,0 +1,108 @@ |
|||||||
|
/*
|
||||||
|
* |
||||||
|
* Copyright 2016 gRPC authors. |
||||||
|
* |
||||||
|
* 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. |
||||||
|
* |
||||||
|
*/ |
||||||
|
|
||||||
|
#include <grpc/support/port_platform.h> |
||||||
|
|
||||||
|
#include "src/core/lib/gprpp/mpscq.h" |
||||||
|
|
||||||
|
namespace grpc_core { |
||||||
|
|
||||||
|
//
|
||||||
|
// MultiProducerSingleConsumerQueue
|
||||||
|
//
|
||||||
|
|
||||||
|
bool MultiProducerSingleConsumerQueue::Push(Node* node) { |
||||||
|
node->next.Store(nullptr, MemoryOrder::RELAXED); |
||||||
|
Node* prev = head_.Exchange(node, MemoryOrder::ACQ_REL); |
||||||
|
prev->next.Store(node, MemoryOrder::RELEASE); |
||||||
|
return prev == &stub_; |
||||||
|
} |
||||||
|
|
||||||
|
MultiProducerSingleConsumerQueue::Node* |
||||||
|
MultiProducerSingleConsumerQueue::Pop() { |
||||||
|
bool empty; |
||||||
|
return PopAndCheckEnd(&empty); |
||||||
|
} |
||||||
|
|
||||||
|
MultiProducerSingleConsumerQueue::Node* |
||||||
|
MultiProducerSingleConsumerQueue::PopAndCheckEnd(bool* empty) { |
||||||
|
Node* tail = tail_; |
||||||
|
Node* next = tail_->next.Load(MemoryOrder::ACQUIRE); |
||||||
|
if (tail == &stub_) { |
||||||
|
// indicates the list is actually (ephemerally) empty
|
||||||
|
if (next == nullptr) { |
||||||
|
*empty = true; |
||||||
|
return nullptr; |
||||||
|
} |
||||||
|
tail_ = next; |
||||||
|
tail = next; |
||||||
|
next = tail->next.Load(MemoryOrder::ACQUIRE); |
||||||
|
} |
||||||
|
if (next != nullptr) { |
||||||
|
*empty = false; |
||||||
|
tail_ = next; |
||||||
|
return tail; |
||||||
|
} |
||||||
|
Node* head = head_.Load(MemoryOrder::ACQUIRE); |
||||||
|
if (tail != head) { |
||||||
|
*empty = false; |
||||||
|
// indicates a retry is in order: we're still adding
|
||||||
|
return nullptr; |
||||||
|
} |
||||||
|
Push(&stub_); |
||||||
|
next = tail->next.Load(MemoryOrder::ACQUIRE); |
||||||
|
if (next != nullptr) { |
||||||
|
*empty = false; |
||||||
|
tail_ = next; |
||||||
|
return tail; |
||||||
|
} |
||||||
|
// indicates a retry is in order: we're still adding
|
||||||
|
*empty = false; |
||||||
|
return nullptr; |
||||||
|
} |
||||||
|
|
||||||
|
//
|
||||||
|
// LockedMultiProducerSingleConsumerQueue
|
||||||
|
//
|
||||||
|
|
||||||
|
bool LockedMultiProducerSingleConsumerQueue::Push(Node* node) { |
||||||
|
return queue_.Push(node); |
||||||
|
} |
||||||
|
|
||||||
|
LockedMultiProducerSingleConsumerQueue::Node* |
||||||
|
LockedMultiProducerSingleConsumerQueue::TryPop() { |
||||||
|
if (gpr_mu_trylock(mu_.get())) { |
||||||
|
Node* node = queue_.Pop(); |
||||||
|
gpr_mu_unlock(mu_.get()); |
||||||
|
return node; |
||||||
|
} |
||||||
|
return nullptr; |
||||||
|
} |
||||||
|
|
||||||
|
LockedMultiProducerSingleConsumerQueue::Node* |
||||||
|
LockedMultiProducerSingleConsumerQueue::Pop() { |
||||||
|
MutexLock lock(&mu_); |
||||||
|
bool empty = false; |
||||||
|
Node* node; |
||||||
|
do { |
||||||
|
node = queue_.PopAndCheckEnd(&empty); |
||||||
|
} while (node == nullptr && !empty); |
||||||
|
return node; |
||||||
|
} |
||||||
|
|
||||||
|
} // namespace grpc_core
|
||||||
@ -0,0 +1,98 @@ |
|||||||
|
/*
|
||||||
|
* |
||||||
|
* Copyright 2016 gRPC authors. |
||||||
|
* |
||||||
|
* 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. |
||||||
|
* |
||||||
|
*/ |
||||||
|
|
||||||
|
#ifndef GRPC_CORE_LIB_GPRPP_MPSCQ_H |
||||||
|
#define GRPC_CORE_LIB_GPRPP_MPSCQ_H |
||||||
|
|
||||||
|
#include <grpc/support/port_platform.h> |
||||||
|
|
||||||
|
#include "src/core/lib/gprpp/atomic.h" |
||||||
|
#include "src/core/lib/gprpp/sync.h" |
||||||
|
|
||||||
|
#include <grpc/support/log.h> |
||||||
|
|
||||||
|
namespace grpc_core { |
||||||
|
|
||||||
|
// Multiple-producer single-consumer lock free queue, based upon the
|
||||||
|
// implementation from Dmitry Vyukov here:
|
||||||
|
// http://www.1024cores.net/home/lock-free-algorithms/queues/intrusive-mpsc-node-based-queue
|
||||||
|
class MultiProducerSingleConsumerQueue { |
||||||
|
public: |
||||||
|
// List node. Application node types can inherit from this.
|
||||||
|
struct Node { |
||||||
|
Atomic<Node*> next; |
||||||
|
}; |
||||||
|
|
||||||
|
MultiProducerSingleConsumerQueue() : head_{&stub_}, tail_(&stub_) {} |
||||||
|
~MultiProducerSingleConsumerQueue() { |
||||||
|
GPR_ASSERT(head_.Load(MemoryOrder::RELAXED) == &stub_); |
||||||
|
GPR_ASSERT(tail_ == &stub_); |
||||||
|
} |
||||||
|
|
||||||
|
// Push a node
|
||||||
|
// Thread safe - can be called from multiple threads concurrently
|
||||||
|
// Returns true if this was possibly the first node (may return true
|
||||||
|
// sporadically, will not return false sporadically)
|
||||||
|
bool Push(Node* node); |
||||||
|
// Pop a node (returns NULL if no node is ready - which doesn't indicate that
|
||||||
|
// the queue is empty!!)
|
||||||
|
// Thread compatible - can only be called from one thread at a time
|
||||||
|
Node* Pop(); |
||||||
|
// Pop a node; sets *empty to true if the queue is empty, or false if it is
|
||||||
|
// not.
|
||||||
|
Node* PopAndCheckEnd(bool* empty); |
||||||
|
|
||||||
|
private: |
||||||
|
// make sure head & tail don't share a cacheline
|
||||||
|
union { |
||||||
|
char padding_[GPR_CACHELINE_SIZE]; |
||||||
|
Atomic<Node*> head_; |
||||||
|
}; |
||||||
|
Node* tail_; |
||||||
|
Node stub_; |
||||||
|
}; |
||||||
|
|
||||||
|
// An mpscq with a lock: it's safe to pop from multiple threads, but doing
|
||||||
|
// only one thread will succeed concurrently.
|
||||||
|
class LockedMultiProducerSingleConsumerQueue { |
||||||
|
public: |
||||||
|
typedef MultiProducerSingleConsumerQueue::Node Node; |
||||||
|
|
||||||
|
// Push a node
|
||||||
|
// Thread safe - can be called from multiple threads concurrently
|
||||||
|
// Returns true if this was possibly the first node (may return true
|
||||||
|
// sporadically, will not return false sporadically)
|
||||||
|
bool Push(Node* node); |
||||||
|
|
||||||
|
// Pop a node (returns NULL if no node is ready - which doesn't indicate that
|
||||||
|
// the queue is empty!!)
|
||||||
|
// Thread safe - can be called from multiple threads concurrently
|
||||||
|
Node* TryPop(); |
||||||
|
|
||||||
|
// Pop a node. Returns NULL only if the queue was empty at some point after
|
||||||
|
// calling this function
|
||||||
|
Node* Pop(); |
||||||
|
|
||||||
|
private: |
||||||
|
MultiProducerSingleConsumerQueue queue_; |
||||||
|
Mutex mu_; |
||||||
|
}; |
||||||
|
|
||||||
|
} // namespace grpc_core
|
||||||
|
|
||||||
|
#endif /* GRPC_CORE_LIB_GPRPP_MPSCQ_H */ |
||||||
Loading…
Reference in new issue