MirrorHub/construct
0
0
Fork 0
mirror of https://github.com/matrix-construct/construct synced 2024-09-29 20:28:52 +02:00
Code Issues Releases Wiki Activity

ircd::ctx: Improve ctx::parallel template.

Browse source
This commit is contained in:
Jason Volk 2018-12-26 17:12:28 -08:00
parent 5fb887096b
commit b0d5e5d4c8
1 changed files with 50 additions and 26 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

76
include/ircd/ctx/parallel.h
View file

@ -26,26 +26,29 @@ struct ircd::ctx::parallel
closure c; closure c;
dock d; dock d;
std::exception_ptr eptr; std::exception_ptr eptr;
size_t snd {0}; ushort snd {0};
size_t rcv {0}; ushort rcv {0};
size_t out {0}; ushort out {0};
void receiver() noexcept;
public: public:
void wait_avail(); void wait_avail();
void wait_done(); void wait_done();
void operator()();
void operator()(const arg &a); void operator()(const arg &a);
parallel(pool &, vector_view<arg>, closure); parallel(pool &, const vector_view<arg> &, closure);
~parallel() noexcept; ~parallel() noexcept;
}; };
template<class arg> template<class arg>
ircd::ctx::parallel<arg>::parallel(pool &p, ircd::ctx::parallel<arg>::parallel(pool &p,
vector_view<arg> a, const vector_view<arg> &a,
closure c) closure c)
:p{&p} :p{&p}
,a{std::move(a)} ,a{a}
,c{std::move(c)} ,c{std::move(c)}
{ {
p.min(this->a.size()); p.min(this->a.size());
@ -67,29 +70,51 @@ ircd::ctx::parallel<arg>::operator()(const arg &a)
if(this->eptr) if(this->eptr)
std::rethrow_exception(this->eptr); std::rethrow_exception(this->eptr);
auto &p(*this->p); this->a.at(snd++) = a;
this->a.at(snd++ % this->a.size()) = a; snd %= this->a.size();
out++; out++;
p([this]()
mutable auto &p(*this->p);
p(std::bind(&parallel::receiver, this));
}
template<class arg>
void
ircd::ctx::parallel<arg>::operator()()
{
wait_avail();
if(this->eptr)
std::rethrow_exception(this->eptr);
snd++;
snd %= this->a.size();
out++;
auto &p(*this->p);
p(std::bind(&parallel::receiver, this));
}
template<class arg>
void
ircd::ctx::parallel<arg>::receiver()
noexcept
{
auto &a
{ {
auto &a this->a.at(rcv++ % this->a.size())
{ };
this->a.at(rcv++ % this->a.size())
};
if(!this->eptr) try if(!this->eptr) try
{ {
c(a); c(a);
} }
catch(...) catch(...)
{ {
this->eptr = std::current_exception(); this->eptr = std::current_exception();
} }
out--; out--;
d.notify_one(); d.notify_one();
});
} }
template<class arg> template<class arg>
@ -98,7 +123,6 @@ ircd::ctx::parallel<arg>::wait_avail()
{ {
d.wait([this] d.wait([this]
{ {
assert(snd >= rcv);
return out < a.size(); return out < a.size();
}); });
} }