MirrorHub/construct
0
0
Fork 0
mirror of https://github.com/matrix-construct/construct synced 2025-01-14 00:34:18 +01:00
Code Issues Releases Wiki Activity

modules/client/sync/rooms/ephemeral/receipt: Add prefetch loops for polylog sync.

Browse source
This commit is contained in:
Jason Volk 2019-09-24 16:06:24 -07:00
parent 0d9d5a8c24
commit b3f45daff8
1 changed files with 60 additions and 17 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

77
modules/client/sync/rooms/ephemeral/receipt.cc
View file

@ -10,10 +10,12 @@
namespace ircd::m::sync
{
extern const m::event::fetch::opts receipt_fopts;
extern conf::item<int64_t> receipt_scan_depth;
static bool _handle_message_receipt(data &, const m::event &);
static bool _handle_message(data &, const m::event::idx &);
static size_t _prefetch_message(data &, const m::event::idx &);
static bool room_ephemeral_m_receipt_m_read_polylog(data &);
static bool room_ephemeral_m_receipt_m_read_linear(data &);
extern item room_ephemeral_m_receipt_m_read;
@ -43,6 +45,12 @@ ircd::m::sync::receipt_scan_depth
{ "default", 10L },
};
decltype(ircd::m::sync::receipt_fopts)
ircd::m::sync::receipt_fopts
{
m::event::keys::include { "content", "sender" }
};
bool
ircd::m::sync::room_ephemeral_m_receipt_m_read_linear(data &data)
{
@ -103,26 +111,66 @@ ircd::m::sync::room_ephemeral_m_receipt_m_read_polylog(data &data)
*data.room
};
// Prefetch loop for event::refs; initial recent messages walk.
ssize_t i(0);
event::idx idx(0);
for(; it && i < receipt_scan_depth; --it)
{
if(apropos(data, it.event_idx()))
{
idx = it.event_idx();
++i;
}
else if(i > 0)
if(!apropos(data, it.event_idx()) && i > 0)
break;
const event::refs refs{it.event_idx()};
refs.prefetch(dbs::ref::M_RECEIPT__M_READ);
idx = it.event_idx();
++i;
}
if(i > 0 && !it && idx)
it.seek(idx);
if(i > 0)
it.seek();
// Prefetch loop for the receipt events
for(ssize_t j(0); it && j < i; --it)
{
if(!apropos(data, it.event_idx()))
continue;
_prefetch_message(data, it.event_idx());
++j;
}
if(i > 0)
it.seek();
// Fetch loop; stream to client.
bool ret{false};
if(i > 0 && idx)
for(; it && i > -1; ++it, --i)
ret |= _handle_message(data, it.event_idx());
for(ssize_t j(0); it && j < i; --it)
{
if(!apropos(data, it.event_idx()))
continue;
ret |= _handle_message(data, it.event_idx());
++j;
}
return ret;
}
size_t
ircd::m::sync::_prefetch_message(data &data,
const event::idx &idx)
{
size_t ret(0);
if(!apropos(data, idx))
return ret;
const event::refs refs{idx};
refs.for_each(dbs::ref::M_RECEIPT__M_READ, [&ret]
(const event::idx &idx, const auto &type)
{
assert(type == dbs::ref::M_RECEIPT__M_READ);
ret += m::prefetch(idx, receipt_fopts);
return true;
});
return ret;
}
@ -140,14 +188,9 @@ ircd::m::sync::_handle_message(data &data,
(const event::idx &idx, const auto &type)
{
assert(type == dbs::ref::M_RECEIPT__M_READ);
static const m::event::fetch::opts fopts
{
m::event::keys::include {"content", "sender"}
};
const m::event::fetch event
{
idx, std::nothrow, fopts
idx, std::nothrow, receipt_fopts
};
if(event.valid)