MirrorHub/construct
0
0
Fork 0
mirror of https://github.com/matrix-construct/construct synced 2024-06-08 13:08:56 +02:00
Code Issues Releases Wiki Activity

modules/media: Eliminate blocks database; encode data into event content. (#139)

Browse source
This commit is contained in:
Jason Volk 2020-08-20 18:56:14 -07:00
parent 472391d5d3
commit 96587f2e6d
4 changed files with 7 additions and 766 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

19
include/ircd/m/media.h
View file

@ -14,6 +14,12 @@
namespace ircd::m::media
{
struct mxc;
extern log::log log;
extern conf::item<size_t> events_prefetch;
extern conf::item<seconds> download_timeout;
extern std::set<m::room::id> downloading;
extern ctx::dock downloading_dock;
}
namespace ircd::m::media::file
@ -44,19 +50,6 @@ namespace ircd::m::media::file
string_view remote = {});
};
namespace ircd::m::media::block
{
using closure = std::function<void (const const_buffer &)>;
bool prefetch(const string_view &hash);
bool get(const string_view &hash, const closure &);
const_buffer get(const mutable_buffer &out, const string_view &hash);
void set(const string_view &hash, const const_buffer &block);
string_view set(const mutable_buffer &hashbuf, const const_buffer &block);
m::event::id::buf set(const room &, const user::id &, const const_buffer &block);
}
struct ircd::m::media::mxc
{
string_view server;

1
matrix/Makefile.am
View file

@ -156,6 +156,7 @@ libircd_matrix_la_SOURCES += gossip.cc
libircd_matrix_la_SOURCES += groups.cc
libircd_matrix_la_SOURCES += request.cc
libircd_matrix_la_SOURCES += keys.cc
libircd_matrix_la_SOURCES += media.cc
libircd_matrix_la_SOURCES += node.cc
libircd_matrix_la_SOURCES += presence.cc
libircd_matrix_la_SOURCES += pretty.cc

729
modules/media/media.cc
View file

@ -14,733 +14,4 @@ ircd::mapi::header
IRCD_MODULE
{
"11.7 :Content respository",
ircd::m::media::init,
ircd::m::media::fini
};
decltype(ircd::m::media::log)
ircd::m::media::log
{
"m.media"
};
decltype(ircd::m::media::blocks_cache_enable)
ircd::m::media::blocks_cache_enable
{
{ "name", "ircd.media.blocks.cache.enable" },
{ "default", true },
};
decltype(ircd::m::media::blocks_cache_comp_enable)
ircd::m::media::blocks_cache_comp_enable
{
{ "name", "ircd.media.blocks.cache_comp.enable" },
{ "default", false },
};
// Blocks column
decltype(ircd::m::media::blocks_descriptor)
ircd::m::media::blocks_descriptor
{
// name
"blocks",
// explain
R"(
Key-value store of blocks belonging to files. The key is a hash of
the block. The key is plaintext sha256-b58 and the block is binary
up to 32768 bytes.
)",
// typing
{
typeid(string_view), typeid(string_view)
},
{}, // options
{}, // comparaor
{}, // prefix transform
false, // drop column
bool(blocks_cache_enable)? -1 : 0,
bool(blocks_cache_comp_enable)? -1 : 0,
// bloom_bits
0,
// expect hit
true,
// block_size
32_KiB,
// meta block size
512,
// compression
{}, // no compression
// compactor
{},
// compaction priority algorithm
"kOldestSmallestSeqFirst"s,
// target file size
{
2_GiB, // base
1L, // multiplier
},
// max bytes for each level
{},
// compaction_period
60s * 60 * 24 * 42,
};
decltype(ircd::m::media::description)
ircd::m::media::description
{
{ "default" }, // requirement of RocksDB
blocks_descriptor,
};
decltype(ircd::m::media::blocks_cache_size)
ircd::m::media::blocks_cache_size
{
{
{ "name", "ircd.media.blocks.cache.size" },
{ "default", long(64_MiB) },
}, []
{
if(!blocks)
return;
const size_t &value{blocks_cache_size};
db::capacity(db::cache(blocks), value);
}
};
decltype(ircd::m::media::blocks_cache_comp_size)
ircd::m::media::blocks_cache_comp_size
{
{
{ "name", "ircd.media.blocks.cache_comp.size" },
{ "default", long(16_MiB) },
}, []
{
if(!blocks)
return;
const size_t &value{blocks_cache_comp_size};
db::capacity(db::cache_compressed(blocks), value);
}
};
decltype(ircd::m::media::blocks_prefetch)
ircd::m::media::blocks_prefetch
{
{ "name", "ircd.media.file.prefetch.blocks" },
{ "default", 32L },
};
decltype(ircd::m::media::events_prefetch)
ircd::m::media::events_prefetch
{
{ "name", "ircd.media.file.prefetch.events" },
{ "default", 16L },
};
decltype(ircd::m::media::database)
ircd::m::media::database;
decltype(ircd::m::media::blocks)
ircd::m::media::blocks;
decltype(ircd::m::media::downloading)
ircd::m::media::downloading;
decltype(ircd::m::media::downloading_dock)
ircd::m::media::downloading_dock;
//
// init
//
void
ircd::m::media::init()
{
static const std::string dbopts;
database = std::make_shared<db::database>("media", dbopts, description);
blocks = db::column{*database, "blocks"};
// The conf setter callbacks must be manually executed after
// the database was just loaded to set the cache size.
conf::reset("ircd.media.blocks.cache.size");
conf::reset("ircd.media.blocks.cache_comp.size");
}
void
ircd::m::media::fini()
{
// The database close contains pthread_join()'s within RocksDB which
// deadlock under certain conditions when called during a dlclose()
// (i.e static destruction of this module). Therefor we must manually
// close the db here first.
database = std::shared_ptr<db::database>{};
}
//
// media::file
//
ircd::m::room::id::buf
IRCD_MODULE_EXPORT
ircd::m::media::file::download(const mxc &mxc,
const m::user::id &user_id,
const string_view &remote)
{
const m::room::id::buf room_id
{
file::room_id(mxc)
};
if(remote && my_host(remote))
return room_id;
if(!remote && my_host(mxc.server))
return room_id;
download(mxc, user_id, room_id, remote);
return room_id;
}
ircd::m::room
IRCD_MODULE_EXPORT
ircd::m::media::file::download(const mxc &mxc,
const m::user::id &user_id,
const m::room::id &room_id,
string_view remote)
try
{
auto iit
{
downloading.emplace(room_id)
};
if(!iit.second)
{
downloading_dock.wait([&room_id]
{
return !downloading.count(room_id);
});
return room_id;
}
const unwind uw{[&iit]
{
downloading.erase(iit.first);
downloading_dock.notify_all();
}};
if(exists(room_id))
return room_id;
if(!remote)
remote = mxc.server;
const unique_buffer<mutable_buffer> buf
{
16_KiB
};
const auto pair
{
download(buf, mxc, remote)
};
const auto &head
{
pair.first
};
const const_buffer &content
{
pair.second
};
char mime_type_buf[64];
const auto &content_type
{
magic::mime(mime_type_buf, content)
};
if(content_type != head.content_type) log::dwarning
{
log, "Server %s claims thumbnail %s is '%s' but we think it is '%s'",
remote,
mxc.mediaid,
head.content_type,
content_type,
};
m::vm::copts vmopts;
const m::room room
{
room_id, &vmopts
};
create(room, user_id, "file");
const unwind_exceptional purge{[&room]
{
m::room::purge(room);
}};
const size_t written
{
file::write(room, user_id, content, content_type)
};
return room;
}
catch(const ircd::server::unavailable &e)
{
throw m::error
{
http::BAD_GATEWAY, "M_MEDIA_UNAVAILABLE",
"Server '%s' is not available for media for '%s/%s' :%s",
remote,
mxc.server,
mxc.mediaid,
e.what()
};
}
decltype(ircd::m::media::download_timeout)
ircd::m::media::download_timeout
{
{ "name", "ircd.media.download.timeout" },
{ "default", 30L },
};
std::pair
<
ircd::http::response::head,
ircd::unique_buffer<ircd::mutable_buffer>
>
IRCD_MODULE_EXPORT
ircd::m::media::file::download(const mutable_buffer &buf_,
const mxc &mxc,
string_view remote,
server::request::opts *const opts)
{
assert(remote || !my_host(mxc.server));
assert(!remote || !my_host(remote));
mutable_buffer buf{buf_};
fed::request::opts fedopts;
fedopts.remote = remote?: mxc.server;
json::get<"method"_>(fedopts.request) = "GET";
json::get<"uri"_>(fedopts.request) = fmt::sprintf
{
buf, "/_matrix/media/r0/download/%s/%s",
mxc.server,
mxc.mediaid,
};
consume(buf, size(json::get<"uri"_>(fedopts.request)));
//TODO: --- This should use the progress callback to build blocks
fed::request remote_request
{
buf, std::move(fedopts)
};
if(!remote_request.wait(seconds(download_timeout), std::nothrow))
throw m::error
{
http::GATEWAY_TIMEOUT, "M_MEDIA_DOWNLOAD_TIMEOUT",
"Server '%s' did not respond with media for '%s/%s' in time",
remote,
mxc.server,
mxc.mediaid
};
const auto &code
{
remote_request.get()
};
if(code != http::OK)
return {};
parse::buffer pb{remote_request.in.head};
parse::capstan pc{pb};
pc.read += size(remote_request.in.head);
return std::pair<http::response::head, unique_buffer<mutable_buffer>>
{
pc, std::move(remote_request.in.dynamic)
};
}
size_t
IRCD_MODULE_EXPORT
ircd::m::media::file::write(const m::room &room,
const m::user::id &user_id,
const const_buffer &content,
const string_view &content_type)
{
//TODO: TXN
send(room, user_id, "ircd.file.stat", "size", json::members
{
{ "value", long(size(content)) }
});
//TODO: TXN
send(room, user_id, "ircd.file.stat", "type", json::members
{
{ "value", content_type }
});
size_t off{0}, wrote{0};
while(off < size(content))
{
const size_t blksz
{
std::min(size(content) - off, size_t(32_KiB))
};
const const_buffer &block
{
data(content) + off, blksz
};
block::set(room, user_id, block);
wrote += size(block);
off += blksz;
}
assert(off == size(content));
assert(wrote == off);
return wrote;
}
size_t
IRCD_MODULE_EXPORT
ircd::m::media::file::read(const m::room &room,
const closure &closure)
{
static const event::fetch::opts fopts
{
event::keys::include { "content", "type" }
};
size_t ret{0};
room::events it
{
room, 1, &fopts
};
if(!it)
return ret;
size_t events_fetched(0), events_prefetched(0);
room::events epf
{
room, 1, &fopts
};
size_t blocks_fetched(0), blocks_prefetched(0);
room::events bpf
{
room, 1, &fopts
};
for(; it; ++it)
{
for(; bpf && blocks_prefetched < blocks_fetched + blocks_prefetch; ++bpf)
{
for(; epf && events_prefetched < events_fetched + events_prefetch; ++epf)
events_prefetched += epf.prefetch();
++events_fetched;
const m::event &event
{
*bpf
};
if(at<"type"_>(event) != "ircd.file.block")
continue;
const json::string &hash
{
at<"content"_>(event).at("hash")
};
blocks_prefetched += block::prefetch(hash);
}
if(!blocks_fetched)
ctx::yield();
++blocks_fetched;
const m::event &event
{
*it
};
if(at<"type"_>(event) != "ircd.file.block")
continue;
const json::string &hash
{
at<"content"_>(event).at("hash")
};
const auto &block_size
{
at<"content"_>(event).get<size_t>("size")
};
const auto handle{[&](const const_buffer &block)
{
if(unlikely(size(block) != block_size))
throw m::NOT_FOUND
{
"File [%s] block [%s] event %s idx:%lu block size %zu != %zu",
string_view{room.room_id},
hash,
string_view{event.event_id},
it.event_idx(),
block_size,
size(block)
};
assert(size(block) == block_size);
ret += size(block);
#if 0
log::debug
{
log, "File %s read %s block[fetched:%zu prefetched:%zu] events[fetched:%zu prefetched:%zu] size:%zu total:%zu",
string_view{room.room_id},
hash,
blocks_fetched,
blocks_prefetched,
events_fetched,
events_prefetched,
block_size,
ret,
};
#endif
closure(block);
}};
if(unlikely(!block::get(hash, handle)))
throw m::NOT_FOUND
{
"File [%s] block %s missing in event %s idx:%lu",
string_view{room.room_id},
hash,
string_view{event.event_id},
it.event_idx(),
};
}
return ret;
}
//
// media::file
//
ircd::m::room::id::buf
IRCD_MODULE_EXPORT
ircd::m::media::file::room_id(const mxc &mxc)
{
m::room::id::buf ret;
room_id(ret, mxc);
return ret;
}
ircd::m::room::id
IRCD_MODULE_EXPORT
ircd::m::media::file::room_id(room::id::buf &out,
const mxc &mxc)
{
thread_local char buf[512];
const auto path
{
mxc.path(buf)
};
const sha256::buf hash
{
sha256{path}
};
out =
{
b58::encode(buf, hash), my_host()
};
return out;
}
//
// media::block
//
ircd::m::event::id::buf
IRCD_MODULE_EXPORT
ircd::m::media::block::set(const m::room &room,
const m::user::id &user_id,
const const_buffer &block)
{
static constexpr const auto bufsz
{
b58::encode_size(sha256::digest_size)
};
char b58buf[bufsz];
const auto hash
{
set(mutable_buffer{b58buf}, block)
};
return send(room, user_id, "ircd.file.block", json::members
{
{ "size", long(size(block)) },
{ "hash", hash }
});
}
ircd::string_view
IRCD_MODULE_EXPORT
ircd::m::media::block::set(const mutable_buffer &b58buf,
const const_buffer &block)
{
const sha256::buf hash
{
sha256{block}
};
const string_view b58hash
{
b58::encode(b58buf, hash)
};
set(b58hash, block);
return b58hash;
}
void
IRCD_MODULE_EXPORT
ircd::m::media::block::set(const string_view &b58hash,
const const_buffer &block)
{
db::write(blocks, b58hash, block);
}
ircd::const_buffer
IRCD_MODULE_EXPORT
ircd::m::media::block::get(const mutable_buffer &out,
const string_view &b58hash)
{
return db::read(blocks, b58hash, out);
}
bool
IRCD_MODULE_EXPORT
ircd::m::media::block::get(const string_view &b58hash,
const closure &closure)
{
const db::gopts opts;
return blocks(b58hash, std::nothrow, closure, opts);
}
bool
IRCD_MODULE_EXPORT
ircd::m::media::block::prefetch(const string_view &b58hash)
{
return db::prefetch(blocks, b58hash);
}
//
// media::mxc
//
IRCD_MODULE_EXPORT
ircd::m::media::mxc::mxc(const string_view &server,
const string_view &mediaid)
:server
{
split(lstrip(server, "mxc://"), '/').first
}
,mediaid
{
mediaid?: rsplit(server, '/').second
}
{
if(empty(server))
throw m::BAD_REQUEST
{
"Invalid MXC: missing server parameter."
};
if(empty(mediaid))
throw m::BAD_REQUEST
{
"Invalid MXC: missing mediaid parameter."
};
}
IRCD_MODULE_EXPORT
ircd::m::media::mxc::mxc(const string_view &uri)
:server
{
split(lstrip(uri, "mxc://"), '/').first
}
,mediaid
{
rsplit(uri, '/').second
}
{
if(empty(server))
throw m::BAD_REQUEST
{
"Invalid MXC: missing server parameter."
};
if(empty(mediaid))
throw m::BAD_REQUEST
{
"Invalid MXC: missing mediaid parameter."
};
}
ircd::string_view
IRCD_MODULE_EXPORT
ircd::m::media::mxc::uri(const mutable_buffer &out)
const
{
return fmt::sprintf
{
out, "mxc://%s/%s", server, mediaid
};
}
ircd::string_view
IRCD_MODULE_EXPORT
ircd::m::media::mxc::path(const mutable_buffer &out)
const
{
return fmt::sprintf
{
out, "%s/%s", server, mediaid
};
}

24
modules/media/media.h
View file

@ -8,30 +8,6 @@
// copyright notice and this permission notice is present in all copies. The
// full license for this software is available in the LICENSE file.
namespace ircd::m::media
{
struct magick;
static void init();
static void fini();
extern log::log log;
extern conf::item<bool> blocks_cache_enable;
extern conf::item<bool> blocks_cache_comp_enable;
extern conf::item<size_t> blocks_cache_size;
extern conf::item<size_t> blocks_cache_comp_size;
extern conf::item<size_t> blocks_prefetch;
extern conf::item<size_t> events_prefetch;
extern const db::descriptor blocks_descriptor;
extern const db::description description;
extern std::shared_ptr<db::database> database;
extern db::column blocks;
extern conf::item<seconds> download_timeout;
extern std::set<m::room::id> downloading;
extern ctx::dock downloading_dock;
}
namespace ircd::m::media::thumbnail
{
extern conf::item<bool> enable;