folly无锁队列，尝试添加新的函数

/*
* Copyright 2014-present Facebook, Inc.
*
* 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.
*/#pragma once#include <atomic>
#include <cassert>
#include <utility>namespace folly { /**
* A very simple atomic single-linked list primitive.
*
* Usage:
*
* class MyClass {
* AtomicIntrusiveLinkedListHook<MyClass> hook_;
* }
*
* AtomicIntrusiveLinkedList<MyClass, &MyClass::hook_> list;
* list.insert(&a);
* list.sweep([] (MyClass* c) { doSomething(c); }
*/
template <class T>
struct AtomicIntrusiveLinkedListHook {
T* next{ nullptr };
}; template <class T, AtomicIntrusiveLinkedListHook<T> T::*HookMember>
class AtomicIntrusiveLinkedList {
public:
AtomicIntrusiveLinkedList() {}
AtomicIntrusiveLinkedList(const AtomicIntrusiveLinkedList&) = delete;
AtomicIntrusiveLinkedList& operator=(const AtomicIntrusiveLinkedList&) =
delete;
AtomicIntrusiveLinkedList(AtomicIntrusiveLinkedList&& other) noexcept {
auto tmp = other.head_.load();
other.head_ = head_.load();
head_ = tmp;
}
AtomicIntrusiveLinkedList& operator=(
AtomicIntrusiveLinkedList&& other) noexcept {
auto tmp = other.head_.load();
other.head_ = head_.load();
head_ = tmp; return *this;
} /**
* Note: list must be empty on destruction.
*/
~AtomicIntrusiveLinkedList() {
assert(empty());
} bool empty() const {
return head_.load() == nullptr;
} /**
* Atomically insert t at the head of the list.
* @return True if the inserted element is the only one in the list
* after the call.
*/
bool insertHead(T* t) {
assert(next(t) == nullptr); auto oldHead = head_.load(std::memory_order_relaxed);
do {
next(t) = oldHead;
/* oldHead is updated by the call below.
NOTE: we don't use next(t) instead of oldHead directly due to
compiler bugs (GCC prior to 4.8.3 (bug 60272), clang (bug 18899),
MSVC (bug 819819); source:
http://en.cppreference.com/w/cpp/atomic/atomic/compare_exchange */
} while (!head_.compare_exchange_weak(oldHead, t,
std::memory_order_release,
std::memory_order_relaxed)); return oldHead == nullptr;
} /**
* Replaces the head with nullptr,
* and calls func() on the removed elements in the order from tail to head.
* Returns false if the list was empty.
*/
template <typename F>
bool sweepOnce(F&& func) {
if (auto head = head_.exchange(nullptr)) {
auto rhead = reverse(head);
unlinkAll(rhead, std::forward<F>(func));
return true;
}
return false;
} // 新增函数
// if std::memory_order_acquire applies to next(oldHead)(the first one, the argument of compare_exchange_weak)
// and I don't know if following bugs affect the code
// GCC prior to 4.8.3 (bug 60272), clang prior to 2014-05-05 (bug 18899)
// MSVC prior to 2014-03-17 (bug 819819).
template <typename F>
bool sweepHead(F&& func)
{
// handle if the list is not empty
auto oldHead = head_.load(std::memory_order_relaxed); while (oldHead != nullptr &&
!head_.compare_exchange_weak(oldHead, next(oldHead),
std::memory_order_acquire, std::memory_order_relaxed))
;
// if drop out head successfully
if (oldHead)
{
next(oldHead) = nullptr;
unlinkAll(oldHead, std::forward<F>(func));
return true;
} return false;
} // 新增函数
// if std::memory_order_acquire does not apply to next(oldHead)
// and I don't know if following bugs affect the code
// GCC prior to 4.8.3 (bug 60272), clang prior to 2014-05-05 (bug 18899)
// MSVC prior to 2014-03-17 (bug 819819).
template <typename F>
bool dropHead(F&& func)
{
T* oldHead = nullptr;
// handle if the list is not empty
while ((oldHead = head_.load(std::memory_order_acquire)))
{
// because insert and drop out will be involving with head_, they
// will change head_ first, then others
bool res = head_.compare_exchange_weak(oldHead, next(oldHead), std::memory_order_relaxed,
std::memory_order_relaxed);
if (res/* && oldHead != nullptr*/)
{
next(oldHead) = nullptr;
unlinkAll(oldHead, std::forward<F>(func));
return true;
}
} return false;
} /**
* Repeatedly replaces the head with nullptr,
* and calls func() on the removed elements in the order from tail to head.
* Stops when the list is empty.
*/
template <typename F>
void sweep(F&& func) {
while (sweepOnce(func)) {
}
} /**
* Similar to sweep() but calls func() on elements in LIFO order.
*
* func() is called for all elements in the list at the moment
* reverseSweep() is called. Unlike sweep() it does not loop to ensure the
* list is empty at some point after the last invocation. This way callers
* can reason about the ordering: elements inserted since the last call to
* reverseSweep() will be provided in LIFO order.
*
* Example: if elements are inserted in the order 1-2-3, the callback is
* invoked 3-2-1. If the callback moves elements onto a stack, popping off
* the stack will produce the original insertion order 1-2-3.
*/
template <typename F>
void reverseSweep(F&& func) {
// We don't loop like sweep() does because the overall order of callbacks
// would be strand-wise LIFO which is meaningless to callers.
auto head = head_.exchange(nullptr);
unlinkAll(head, std::forward<F>(func));
} private:
std::atomic<T*> head_{ nullptr }; static T*& next(T* t) {
return (t->*HookMember).next;
} /* Reverses a linked list, returning the pointer to the new head
(old tail) */
static T* reverse(T* head) {
T* rhead = nullptr;
while (head != nullptr) {
auto t = head;
head = next(t);
next(t) = rhead;
rhead = t;
}
return rhead;
} /* Unlinks all elements in the linked list fragment pointed to by `head',
* calling func() on every element */
template <typename F>
void unlinkAll(T* head, F&& func) {
while (head != nullptr) {
auto t = head;
head = next(t);
next(t) = nullptr;
func(t);
}
}
};} // namespace folly

// Test.cpp : 定义控制台应用程序的入口点。
//#include <memory>
#include <cassert>#include <iostream>
#include <vector>
#include <thread>
#include <future>
#include <random>
#include <cmath>#include "folly.h"using namespace folly;struct student_name
{
//student_name(const std::string& name)
// : name(name)
//{ //} //std::string name;
student_name(int age = )
: age(age)
{ } int age;
AtomicIntrusiveLinkedListHook<student_name> node;
};AtomicIntrusiveLinkedList<student_name, &student_name::node> g_students;std::atomic<int> g_inserts; // insert num (successful)
std::atomic<int> g_drops; // drop num (successful)std::atomic<int> g_printNum; // as same as g_dropsstd::atomic<long> g_ageInSum; // age sum when producing student_name
std::atomic<long> g_ageOutSum; // age sum when consuming student_namestd::atomic<bool> goOn(true);constexpr int ONE_THREAD_PRODUCE_NUM = ; // when testing, no more than this number, you know 20,000,00 * 100 * 10~= MAX_INTconstexpr int PRODUCE_THREAD_NUM = ; // producing thread number
constexpr int CONSUME_THREAD_NUM = ; // consuming thread numbervoid printOne(student_name* t)
{
g_printNum.fetch_add(, std::memory_order_relaxed);
g_ageOutSum.fetch_add(t->age, std::memory_order_relaxed);
// use memory_order_relaxed avoiding affect folly memory order
g_drops.fetch_add(, std::memory_order_relaxed);
// clean node
// delete t;
}void insert_students()
{
std::default_random_engine dre(time(nullptr));
std::uniform_int_distribution<int> ageDi(, ); for (int i = ; i != ONE_THREAD_PRODUCE_NUM; ++i)
{
int newAge = ageDi(dre);
g_ageInSum.fetch_add(newAge, std::memory_order_relaxed);
g_students.insertHead(new student_name(newAge));
// use memory_order_relaxed avoiding affect folly memory order
g_inserts.fetch_add(, std::memory_order_relaxed);
}
}void drop_students()
{
while (goOn.load(std::memory_order_relaxed))
{
g_students.dropHead(printOne);
}
}int main()
{
std::vector<std::future<void>> insert_threads;
for (int i = ; i != PRODUCE_THREAD_NUM; ++i)
{
insert_threads.push_back(std::async(insert_students));
} std::vector<std::future<void>> drop_threads;
for (int i = ; i != CONSUME_THREAD_NUM; ++i)
{
drop_threads.push_back(std::async(drop_students));
} for (auto& item : insert_threads)
{
item.get();
} goOn.store(std::memory_order_relaxed); for (auto& item : drop_threads)
{
item.get();
} g_students.sweep(printOne); std::cout << "insert count1: " << g_inserts.load() << std::endl;
std::cout << "drop count1: " << g_drops.load() << std::endl;
std::cout << "print num1: " << g_printNum.load() << std::endl; std::cout << "age in1: " << g_ageInSum.load() << std::endl;
std::cout << "age out1: " << g_ageOutSum.load() << std::endl; std::cout << std::endl;
}