1
2 // Copyright Oliver Kowalke 2015.
3 // Distributed under the Boost Software License, Version 1.0.
4 // (See accompanying file LICENSE_1_0.txt or copy at
5 // http://www.boost.org/LICENSE_1_0.txt)
6
7 // based on https://github.com/atemerev/skynet from Alexander Temerev
8
9 #include <algorithm>
10 #include <cassert>
11 #include <chrono>
12 #include <cmath>
13 #include <condition_variable>
14 #include <cstddef>
15 #include <cstdint>
16 #include <cstdlib>
17 #include <queue>
18 #include <iostream>
19 #include <memory>
20 #include <mutex>
21 #include <numeric>
22 #include <random>
23 #include <sstream>
24 #include <vector>
25
26 #include <boost/fiber/all.hpp>
27 #include <boost/predef.h>
28
29 #include "barrier.hpp"
30
31 using clock_type = std::chrono::steady_clock;
32 using duration_type = clock_type::duration;
33 using time_point_type = clock_type::time_point;
34 using channel_type = boost::fibers::buffered_channel< std::uint64_t >;
35 using allocator_type = boost::fibers::fixedsize_stack;
36 using lock_type = std::unique_lock< std::mutex >;
37
38 static bool done = false;
39 static std::mutex mtx{};
40 static boost::fibers::condition_variable_any cnd{};
41
42 // microbenchmark
skynet(allocator_type & salloc,std::uint64_t num,std::uint64_t size,std::uint64_t div)43 std::uint64_t skynet(allocator_type& salloc, std::uint64_t num, std::uint64_t size, std::uint64_t div) {
44 if ( size != 1){
45 size /= div;
46
47 std::vector<boost::fibers::future<std::uint64_t> > results;
48 results.reserve( div);
49
50 for ( std::uint64_t i = 0; i != div; ++i) {
51 std::uint64_t sub_num = num + i * size;
52 results.emplace_back(boost::fibers::async(
53 boost::fibers::launch::dispatch
54 , std::allocator_arg, salloc
55 , skynet
56 , std::ref( salloc), sub_num, size, div));
57 }
58
59 std::uint64_t sum = 0;
60 for ( auto& f : results)
61 sum += f.get();
62
63 return sum;
64 }
65
66 return num;
67 }
68
thread(std::uint32_t thread_count)69 void thread( std::uint32_t thread_count) {
70 // thread registers itself at work-stealing scheduler
71 boost::fibers::use_scheduling_algorithm< boost::fibers::algo::work_stealing >( thread_count);
72 lock_type lk( mtx);
73 cnd.wait( lk, [](){ return done; });
74 BOOST_ASSERT( done);
75 }
76
main()77 int main() {
78 try {
79 // count of logical ids
80 std::uint32_t thread_count = std::thread::hardware_concurrency();
81 std::size_t size{ 1000000 };
82 std::size_t div{ 10 };
83 allocator_type salloc{ 2*allocator_type::traits_type::page_size() };
84 std::uint64_t result{ 0 };
85 channel_type rc{ 2 };
86 std::vector< std::thread > threads;
87 for ( std::uint32_t i = 1 /* count main-thread */; i < thread_count; ++i) {
88 // spawn thread
89 threads.emplace_back( thread, thread_count);
90 }
91 // main-thread registers itself at work-stealing scheduler
92 boost::fibers::use_scheduling_algorithm< boost::fibers::algo::work_stealing >( thread_count);
93 time_point_type start{ clock_type::now() };
94 result = skynet( salloc, 0, size, div);
95 if ( 499999500000 != result) {
96 throw std::runtime_error("invalid result");
97 }
98 auto duration = clock_type::now() - start;
99 lock_type lk( mtx);
100 done = true;
101 lk.unlock();
102 cnd.notify_all();
103 for ( std::thread & t : threads) {
104 t.join();
105 }
106 std::cout << "duration: " << duration.count() / 1000000 << " ms" << std::endl;
107 return EXIT_SUCCESS;
108 } catch ( std::exception const& e) {
109 std::cerr << "exception: " << e.what() << std::endl;
110 } catch (...) {
111 std::cerr << "unhandled exception" << std::endl;
112 }
113 return EXIT_FAILURE;
114 }
115