Merge pull request #28 from Microsoft/mpscq

Updated to match documentation.
This commit is contained in:
Matthew Parkinson
2019-02-19 16:22:00 +00:00
committed by GitHub
2 changed files with 27 additions and 42 deletions

View File

@@ -15,85 +15,70 @@ namespace snmalloc
std::is_same<decltype(((T*)0)->next), std::atomic<T*>>::value, std::is_same<decltype(((T*)0)->next), std::atomic<T*>>::value,
"T->next must be a std::atomic<T*>"); "T->next must be a std::atomic<T*>");
std::atomic<T*> head; std::atomic<T*> back;
T* tail; T* front;
public: public:
void invariant() void invariant()
{ {
#ifndef NDEBUG #ifndef NDEBUG
assert(head != nullptr); assert(back != nullptr);
assert(tail != nullptr); assert(front != nullptr);
#endif #endif
} }
void init(T* stub) void init(T* stub)
{ {
stub->next.store(nullptr, std::memory_order_relaxed); stub->next.store(nullptr, std::memory_order_relaxed);
tail = stub; front = stub;
head.store(stub, std::memory_order_relaxed); back.store(stub, std::memory_order_relaxed);
invariant(); invariant();
} }
T* destroy() T* destroy()
{ {
T* tl = tail; T* fnt = front;
head.store(nullptr, std::memory_order_relaxed); back.store(nullptr, std::memory_order_relaxed);
tail = nullptr; front = nullptr;
return tl; return fnt;
}
T* get_head()
{
return head.load(std::memory_order_relaxed);
}
inline void push(T* item)
{
push(item, item);
} }
inline bool is_empty() inline bool is_empty()
{ {
T* hd = head.load(std::memory_order_relaxed); T* bk = back.load(std::memory_order_relaxed);
return hd == tail; return bk == front;
} }
void push(T* first, T* last) void enqueue(T* first, T* last)
{ {
// Pushes a list of messages to the queue. Each message from first to // Pushes a list of messages to the queue. Each message from first to
// last should be linked together through their next pointers. // last should be linked together through their next pointers.
invariant(); invariant();
last->next.store(nullptr, std::memory_order_relaxed); last->next.store(nullptr, std::memory_order_relaxed);
std::atomic_thread_fence(std::memory_order_release); std::atomic_thread_fence(std::memory_order_release);
T* prev = head.exchange(last, std::memory_order_relaxed); T* prev = back.exchange(last, std::memory_order_relaxed);
prev->next.store(first, std::memory_order_relaxed); prev->next.store(first, std::memory_order_relaxed);
} }
std::pair<T*, T*> pop() T* dequeue()
{ {
// Returns the next message and the tail message. If the next message // Returns the front message, or null if not possible to return a message.
// is not null, the tail message should be freed by the caller.
invariant(); invariant();
T* tl = tail; T* first = front;
T* next = tl->next.load(std::memory_order_relaxed); T* next = first->next.load(std::memory_order_relaxed);
if (next != nullptr) if (next != nullptr)
{ {
tail = next; front = next;
assert(tail); assert(front);
std::atomic_thread_fence(std::memory_order_acquire); std::atomic_thread_fence(std::memory_order_acquire);
invariant();
return first;
} }
invariant(); return nullptr;
return std::make_pair(next, tl);
}
T* peek()
{
return tail->next.load(std::memory_order_relaxed);
} }
}; };
} }

View File

@@ -583,7 +583,7 @@ namespace snmalloc
{ {
// Send all slots to the target at the head of the list. // Send all slots to the target at the head of the list.
Superslab* super = Superslab::get(first); Superslab* super = Superslab::get(first);
super->get_allocator()->message_queue.push(first, l->last); super->get_allocator()->message_queue.enqueue(first, l->last);
l->clear(); l->clear();
} }
} }
@@ -776,12 +776,12 @@ namespace snmalloc
{ {
for (size_t i = 0; i < REMOTE_BATCH; i++) for (size_t i = 0; i < REMOTE_BATCH; i++)
{ {
std::pair<Remote*, Remote*> r = message_queue().pop(); Remote* r = message_queue().dequeue();
if (std::get<0>(r) == nullptr) if (r == nullptr)
break; break;
handle_dealloc_remote(std::get<1>(r)); handle_dealloc_remote(r);
} }
// Our remote queues may be larger due to forwarding remote frees. // Our remote queues may be larger due to forwarding remote frees.