Skip to content

Commit

Permalink
2015-07-16 19:36:04.108966060+01:00
Browse files Browse the repository at this point in the history
  • Loading branch information
@jenkins-nedprod
jenkins-nedprod committed Jul 16, 2015
1 parent 2587f1a commit a5a286f
Show file tree
Hide file tree
Showing 71 changed files with 14,108 additions and 12,263 deletions.
92 changes: 41 additions & 51 deletions example/adopt_example.cpp
View file
Original file line number Diff line number Diff line change
Expand Up @@ -3,60 +3,50 @@
//[adopt_example
struct test_handle : boost::afio::async_io_handle
{
test_handle(boost::afio::async_file_io_dispatcher_base *parent) :
boost::afio::async_io_handle(parent,
boost::afio::file_flags::None) {}
virtual void close() override final
{
// Do nothing
}
virtual async_io_handle::open_states is_open() const override final
{
return async_io_handle::open_states::open;
}
virtual void *native_handle() const override final
{
return nullptr;
}
virtual boost::afio::path path(bool refresh=false) override final
{
return boost::afio::path();
}
virtual boost::afio::path path() const override final
{
return boost::afio::path();
}
virtual boost::afio::directory_entry direntry(boost::afio::metadata_flags
wanted=boost::afio::directory_entry::metadata_fastpath()) override final
{
return boost::afio::directory_entry();
}
virtual boost::afio::path target() override final
{
return boost::afio::path();
}
virtual void *try_mapfile() override final
{
return nullptr;
}
virtual void link(const boost::afio::async_path_op_req &req) override final
{
}
virtual void unlink() override final
{
}
virtual void atomic_relink(const boost::afio::async_path_op_req &req) override final
{
}
test_handle(boost::afio::async_file_io_dispatcher_base *parent)
: boost::afio::async_io_handle(parent, boost::afio::file_flags::None)
{
}
virtual void close() override final
{
// Do nothing
}
virtual async_io_handle::open_states is_open() const override final
{
return async_io_handle::open_states::open;
}
virtual void *native_handle() const override final { return nullptr; }
virtual boost::afio::path path(bool refresh = false) override final
{
return boost::afio::path();
}
virtual boost::afio::path path() const override final
{
return boost::afio::path();
}
virtual boost::afio::directory_entry
direntry(boost::afio::metadata_flags
wanted = boost::afio::directory_entry::metadata_fastpath()) override final
{
return boost::afio::directory_entry();
}
virtual boost::afio::path target() override final { return boost::afio::path(); }
virtual void *try_mapfile() override final { return nullptr; }
virtual void link(const boost::afio::async_path_op_req &req) override final {}
virtual void unlink() override final {}
virtual void
atomic_relink(const boost::afio::async_path_op_req &req) override final
{
}
};

int main(void)
{
auto dispatcher = boost::afio::make_async_file_io_dispatcher(
boost::afio::process_threadpool());
auto h=std::make_shared<test_handle>(dispatcher.get());
auto adopted=dispatcher->adopt(h);
when_all(adopted).wait();
return 0;
auto dispatcher =
boost::afio::make_async_file_io_dispatcher(boost::afio::process_threadpool());
auto h = std::make_shared<test_handle>(dispatcher.get());
auto adopted = dispatcher->adopt(h);
when_all(adopted).wait();
return 0;
}
//]
125 changes: 66 additions & 59 deletions example/barrier_example.cpp
View file
Original file line number Diff line number Diff line change
Expand Up @@ -2,66 +2,73 @@

int main(void)
{
//[barrier_example
// Assume that groups is 10,000 items long with item.first being randomly
// between 1 and 500. This example is adapted from the barrier() unit test.
//
// What we're going to do is this: for each item in groups, schedule item.first
// parallel ops and a barrier which completes only when the last of that
// parallel group completes. Chain the next group to only execute after the
// preceding group's barrier completes. Repeat until all groups have been executed.
std::shared_ptr<boost::afio::async_file_io_dispatcher_base> dispatcher=
boost::afio::make_async_file_io_dispatcher();
std::vector<std::pair<size_t, int>> groups;
boost::afio::atomic<size_t> callcount[10000];
memset(&callcount, 0, sizeof(callcount));

// This lambda is what each parallel op in each group will do: increment an atomic
// for that group.
auto inccount = [](boost::afio::atomic<size_t> *count){ (*count)++; };

// This lambda is called after each barrier completes, and it checks that exactly
// the right number of inccount lambdas were executed.
auto verifybarrier = [](boost::afio::atomic<size_t> *count, size_t shouldbe)
//[barrier_example
// Assume that groups is 10,000 items long with item.first being randomly
// between 1 and 500. This example is adapted from the barrier() unit test.
//
// What we're going to do is this: for each item in groups, schedule item.first
// parallel ops and a barrier which completes only when the last of that
// parallel group completes. Chain the next group to only execute after the
// preceding group's barrier completes. Repeat until all groups have been executed.
std::shared_ptr<boost::afio::async_file_io_dispatcher_base> dispatcher =
boost::afio::make_async_file_io_dispatcher();
std::vector<std::pair<size_t, int>> groups;
boost::afio::atomic<size_t> callcount[10000];
memset(&callcount, 0, sizeof(callcount));

// This lambda is what each parallel op in each group will do: increment an atomic
// for that group.
auto inccount = [](boost::afio::atomic<size_t> *count)
{
(*count)++;
};

// This lambda is called after each barrier completes, and it checks that exactly
// the right number of inccount lambdas were executed.
auto verifybarrier = [](boost::afio::atomic<size_t> *count, size_t shouldbe)
{
if(*count != shouldbe)
throw std::runtime_error("Count was not what it should have been!");
return true;
};

// For each group, dispatch ops and a barrier for them
boost::afio::async_io_op next;
bool isfirst = true;
for(auto &run : groups)
{
// Create a vector of run.first size of bound inccount lambdas
// This will be the batch issued for this group
std::vector<std::function<void()>> thisgroupcalls(
run.first, std::bind(inccount, &callcount[run.second]));
std::vector<boost::afio::async_io_op> thisgroupcallops;
// If this is the first item, schedule without precondition
if(isfirst)
{
if (*count != shouldbe)
throw std::runtime_error("Count was not what it should have been!");
return true;
};

// For each group, dispatch ops and a barrier for them
boost::afio::async_io_op next;
bool isfirst = true;
for(auto &run : groups)
thisgroupcallops = dispatcher->call(thisgroupcalls).second;
isfirst = false;
}
else
{
// Create a vector of run.first size of bound inccount lambdas
// This will be the batch issued for this group
std::vector<std::function<void()>> thisgroupcalls(run.first, std::bind(inccount, &callcount[run.second]));
std::vector<boost::afio::async_io_op> thisgroupcallops;
// If this is the first item, schedule without precondition
if (isfirst)
{
thisgroupcallops = dispatcher->call(thisgroupcalls).second;
isfirst = false;
}
else
{
// Create a vector of run.first size of preconditions exactly
// matching the number in this batch. Note that the precondition
// for all of these is the preceding verify op
std::vector<boost::afio::async_io_op> dependency(run.first, next);
thisgroupcallops = dispatcher->call(dependency, thisgroupcalls).second;
}
// barrier() is very easy: its number of output ops exactly matches its input
// but none of the output will complete until the last of the input completes
auto thisgroupbarriered = dispatcher->barrier(thisgroupcallops);
// Schedule a call of the verify lambda once barrier completes. Here we choose
// the first item of the barrier's return, but in truth any of them are good.
auto verify = dispatcher->call(thisgroupbarriered.front(), std::function<bool()>(std::bind(verifybarrier, &callcount[run.second], run.first)));
// Set the dependency for the next batch to be the just scheduled verify op
next = verify.second;
// Create a vector of run.first size of preconditions exactly
// matching the number in this batch. Note that the precondition
// for all of these is the preceding verify op
std::vector<boost::afio::async_io_op> dependency(run.first, next);
thisgroupcallops = dispatcher->call(dependency, thisgroupcalls).second;
}
// next was the last op scheduled, so waiting on it waits on everything
when_all(next).wait();
//]
// barrier() is very easy: its number of output ops exactly matches its input
// but none of the output will complete until the last of the input completes
auto thisgroupbarriered = dispatcher->barrier(thisgroupcallops);
// Schedule a call of the verify lambda once barrier completes. Here we choose
// the first item of the barrier's return, but in truth any of them are good.
auto verify =
dispatcher->call(thisgroupbarriered.front(),
std::function<bool()>(
std::bind(verifybarrier, &callcount[run.second], run.first)));
// Set the dependency for the next batch to be the just scheduled verify op
next = verify.second;
}
// next was the last op scheduled, so waiting on it waits on everything
when_all(next).wait();
//]
}
52 changes: 28 additions & 24 deletions example/benchmark_asio.cpp
View file
Original file line number Diff line number Diff line change
Expand Up @@ -10,38 +10,42 @@ static int callback()
#if 0
Sleep(0);
#endif
--togo;
return 1;
--togo;
return 1;
};
int main(void)
{
using namespace boost::afio;
typedef chrono::duration<double, ratio<1, 1>> secs_type;
auto threadpool=process_threadpool();
auto begin=chrono::high_resolution_clock::now();
while(chrono::duration_cast<secs_type>(chrono::high_resolution_clock::now()-begin).count()<3);

atomic<size_t> threads(0);
using namespace boost::afio;
typedef chrono::duration<double, ratio<1, 1>> secs_type;
auto threadpool = process_threadpool();
auto begin = chrono::high_resolution_clock::now();
while(chrono::duration_cast<secs_type>(chrono::high_resolution_clock::now() -
begin).count() < 3)
;

atomic<size_t> threads(0);
#if 0
std::cout << "Attach profiler now and hit Return" << std::endl;
getchar();
#endif
begin=chrono::high_resolution_clock::now();
begin = chrono::high_resolution_clock::now();
#pragma omp parallel
{
++threads;
for(size_t n = 0; n < 5000000; n++)
{
++threads;
for(size_t n=0; n<5000000; n++)
{
++togo;
threadpool->enqueue(callback);
}
++togo;
threadpool->enqueue(callback);
}
while(togo)
this_thread::sleep_for(chrono::milliseconds(1));
auto end=chrono::high_resolution_clock::now();
auto diff=chrono::duration_cast<secs_type>(end-begin);
std::cout << "It took " << diff.count() << " secs to execute " << (5000000*threads) << " closures which is " << (5000000*threads/diff.count()) << " closures/sec" << std::endl;
std::cout << "\nPress Return to exit ..." << std::endl;
getchar();
return 0;
}
while(togo)
this_thread::sleep_for(chrono::milliseconds(1));
auto end = chrono::high_resolution_clock::now();
auto diff = chrono::duration_cast<secs_type>(end - begin);
std::cout << "It took " << diff.count() << " secs to execute "
<< (5000000 * threads) << " closures which is "
<< (5000000 * threads / diff.count()) << " closures/sec" << std::endl;
std::cout << "\nPress Return to exit ..." << std::endl;
getchar();
return 0;
}
Loading

0 comments on commit a5a286f

Please sign in to comment.