p/dawn
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases 1 Activity

Deduplicate IPC RPC

Browse Source
This commit is contained in:
p
2026-09-08 11:39:55 +02:00
parent 0df1955435
commit 7962d9becb
14 changed files with 663 additions and 742 deletions
Download Patch File Download Diff File Expand all files Collapse all files
cmake/lxdr-generate.cmake
+22 -8
View File
@@ -4,34 +4,47 @@
# Copyright The Dawn Authors
# SPDX-License-Identifier: MPL-2.0
#
# lxdr_generate(<out_header> <namespace> <prefix_camel> <lxdr>...)
# lxdr_generate(<out_header> <namespace> <prefix_camel> <lxdr>...
# [EXTERN <lxdr>...])
#
# Writes <out_header> (typically .../<name>.lxdr.hpp) from the listed
# .lxdr files. Does not attach the header to any compile target.
# .lxdr files. EXTERN ones are read for their types only, and included
# from beside <out_header> rather than repeated. Does not attach the
# header to any compile target.
#
function(lxdr_generate out_header namespace prefix_camel)
cmake_parse_arguments(arg "" "" "EXTERN" ${ARGN})
if(NOT AWK)
message(FATAL_ERROR "lxdr_generate requires AWK")
endif()
if(NOT ARGN)
if(NOT arg_UNPARSED_ARGUMENTS)
message(FATAL_ERROR "lxdr_generate: no .lxdr inputs")
endif()
set(lxdrgen "${PROJECT_SOURCE_DIR}/submodules/liberty/tools/lxdrgen.awk")
set(lxdrgen_cpp "${PROJECT_SOURCE_DIR}/ipc/lxdrgen-cpp.awk")
get_filename_component(out_dir "${out_header}" DIRECTORY)
get_filename_component(out_name "${out_header}" NAME)
get_filename_component(out_subdir "${out_dir}" NAME)
string(MAKE_C_IDENTIFIER "${out_name}" out_ident)
# The extern ones have to come first, and are named to AWK by the
# include stem their own header was generated under.
set(lxdr_abs)
foreach(f IN LISTS ARGN)
set(extern_stems)
foreach(f IN LISTS arg_EXTERN arg_UNPARSED_ARGUMENTS)
if(NOT IS_ABSOLUTE "${f}")
get_filename_component(f "${f}" ABSOLUTE)
endif()
list(APPEND lxdr_abs "${f}")
endforeach()
get_filename_component(out_dir "${out_header}" DIRECTORY)
get_filename_component(out_name "${out_header}" NAME)
string(MAKE_C_IDENTIFIER "${out_name}" out_ident)
foreach(f IN LISTS arg_EXTERN)
get_filename_component(f "${f}" NAME)
list(APPEND extern_stems "${out_subdir}/${f}")
endforeach()
list(JOIN extern_stems " " extern_arg)
add_custom_command(
OUTPUT "${out_header}"
@@ -40,6 +53,7 @@ function(lxdr_generate out_header namespace prefix_camel)
${AWK} -f "${lxdrgen}" -f "${lxdrgen_cpp}"
-v "PrefixCamel=${prefix_camel}"
-v "Namespace=${namespace}"
-v "Extern=${extern_arg}"
${lxdr_abs}
> "${out_header}"
DEPENDS ${lxdr_abs} "${lxdrgen}" "${lxdrgen_cpp}"
dn/instance.cpp
+6 -6
View File
@@ -52,21 +52,21 @@ make_watch(dawn::ipc::Waitable w, bool write, QObject *parent)
}
#endif
static dawn::ipc::instance::ErrorCode
static dawn::ipc::ErrorCode
map_open_error(OpenResult r)
{
switch (r) {
case OpenResult::NotFound:
return dawn::ipc::instance::ErrorCode::NotFound;
return dawn::ipc::ErrorCode::NotFound;
case OpenResult::PermissionDenied:
return dawn::ipc::instance::ErrorCode::PermissionDenied;
return dawn::ipc::ErrorCode::PermissionDenied;
case OpenResult::InvalidArgument:
return dawn::ipc::instance::ErrorCode::InvalidArgument;
return dawn::ipc::ErrorCode::InvalidArgument;
case OpenResult::Ok:
case OpenResult::Internal:
break;
}
return dawn::ipc::instance::ErrorCode::Internal;
return dawn::ipc::ErrorCode::Internal;
}
struct InstanceHost::Impl {
@@ -163,7 +163,7 @@ InstanceHost::Impl::on_request(
const auto *open_body =
get_if<dawn::ipc::instance::RequestBodyOpenView>(&req.body.value);
if (!open_body) {
call.fail(dawn::ipc::instance::ErrorCode::InvalidArgument, {});
call.fail(dawn::ipc::ErrorCode::InvalidArgument, {});
return;
}
dn/main.cpp
+3 -3
View File
@@ -63,9 +63,9 @@ instance_session()
}
static const char *
error_fallback(dawn::ipc::instance::ErrorCode code)
error_fallback(dawn::ipc::ErrorCode code)
{
using dawn::ipc::instance::ErrorCode;
using dawn::ipc::ErrorCode;
switch (code) {
case ErrorCode::NotFound:
return "not found";
@@ -96,7 +96,7 @@ handoff_open(
if (const uint32_t pid = client.server_pid())
AllowSetForegroundWindow(DWORD(pid));
#endif
dawn::ipc::instance::Error error;
dawn::ipc::Error error;
if (client.open(encoded, token, browse, &error, dawn::ipc::kRequestTimeout))
return true;
if (!error.message.empty())
dnimaged/main.cpp
+66 -253
View File
@@ -5,17 +5,14 @@
// SPDX-License-Identifier: MPL-2.0
//
#include "ipc-loop.hpp"
#include "ipc-rpc.hpp"
#include "ipc-shm.hpp"
#include "ipc/imaged.lxdr.hpp"
#include "libdn.h"
#include <algorithm>
#include <condition_variable>
#include <cstdio>
#include <deque>
#include <mutex>
#include <memory>
#include <thread>
#include <unordered_set>
#include <variant>
@@ -28,92 +25,41 @@ namespace proto = dawn::ipc::imaged;
static constexpr char kService[] = "imaged";
static constexpr uint64_t kMaxBlob = 1024ull * 1024 * 1024;
// Four decoder workers share a bounded retained-input pool. Per-client
// admission keeps one connection from occupying the whole queue.
static constexpr size_t kMaxJobs = 8, kMaxJobsPerConnection = 2;
static constexpr size_t kMaxRetainedBytes =
1024ull * 1024 * 1024 + dawn::ipc::Connection::kMaxPayload;
static vector<byte>
encode_frame(const proto::Frame &frame)
error_frame(uint64_t id, string message, dawn::ipc::ErrorCode code)
{
vector<byte> out;
dawn::ipc::Encoder encoder(out);
encode(frame, encoder);
return out;
proto::Frame frame;
frame.payload.value = proto::PayloadResponse{proto::Response{id,
proto::Result{
proto::ResultError{dawn::ipc::Error{code, std::move(message)}}}}};
return dawn::ipc::encoded(frame);
}
namespace
{
struct Job {
uint64_t connection = 0, request = 0;
size_t retained_bytes = 0;
uint64_t request = 0;
vector<byte> data, target_icc;
dawn::ipc::SharedMemory shared;
bool first_frame_only = false;
};
struct Done {
uint64_t connection = 0;
proto::Frame frame;
dawn::ipc::SharedMemory memory;
};
struct State {
mutex mu;
condition_variable cv;
deque<Job> jobs;
deque<Done> done;
unordered_multiset<uint64_t> admitted;
unordered_set<uint64_t> greeted;
size_t retained_bytes = 0;
bool stop = false;
dawn::ipc::Loop loop;
dawn::ipc::ServerCore *server = nullptr;
dawn::ipc::DaemonHost *host = nullptr;
};
} // namespace
static bool
admit(State &state, Job &job, size_t bytes)
static dawn::ipc::DaemonHost::Reply
decode_job(const Job &job)
{
lock_guard lock(state.mu);
if (state.admitted.size() >= kMaxJobs ||
state.admitted.count(job.connection) >= kMaxJobsPerConnection ||
bytes > kMaxRetainedBytes ||
state.retained_bytes > kMaxRetainedBytes - bytes)
return false;
// libdn wants a Cmm per thread, and this only ever runs on workers.
static thread_local auto cmm = make_shared<dawn::Cmm>();
state.admitted.insert(job.connection);
state.retained_bytes += bytes;
job.retained_bytes = bytes;
return true;
}
static void
release(State &state, uint64_t connection, size_t bytes)
{
lock_guard lock(state.mu);
state.retained_bytes -= bytes;
state.admitted.erase(state.admitted.find(connection));
}
static proto::Frame
error_frame(uint64_t id, string message, proto::ErrorCode code)
{
proto::Frame frame;
frame.payload.value = proto::PayloadResponse{proto::Response{id,
proto::Result{
proto::ResultError{proto::Error{code, std::move(message)}}}}};
return frame;
}
static Done
decode_job(Job &job, const shared_ptr<dawn::Cmm> &cmm)
{
Done done;
done.connection = job.connection;
dawn::ipc::DaemonHost::Reply reply;
dawn::OpenContext context;
context.cmm = cmm;
context.first_frame_only = job.first_frame_only;
@@ -123,9 +69,9 @@ decode_job(Job &job, const shared_ptr<dawn::Cmm> &cmm)
reinterpret_cast<const uint8_t *>(job.target_icc.data()),
job.target_icc.size()));
if (!context.screen_profile) {
done.frame = error_frame(job.request, "invalid target ICC profile",
proto::ErrorCode::InvalidArgument);
return done;
reply.payload = error_frame(job.request, "invalid target ICC profile",
dawn::ipc::ErrorCode::InvalidArgument);
return reply;
}
dawn::Error error;
@@ -135,15 +81,15 @@ decode_job(Job &job, const shared_ptr<dawn::Cmm> &cmm)
const size_t size = job.shared.ok() ? job.shared.size() : job.data.size();
auto image = dawn::open_from_data(span(data, size), context, &error);
if (!image) {
done.frame = error_frame(job.request,
reply.payload = error_frame(job.request,
error.message.empty() ? "decode failed" : error.message,
proto::ErrorCode::InvalidArgument);
return done;
dawn::ipc::ErrorCode::InvalidArgument);
return reply;
}
if (image->data.empty() || image->data.size() > kMaxBlob) {
done.frame = error_frame(job.request, "decoded pixmap is too large",
proto::ErrorCode::InvalidArgument);
return done;
reply.payload = error_frame(job.request, "decoded pixmap is too large",
dawn::ipc::ErrorCode::InvalidArgument);
return reply;
}
proto::DecodeResponse response;
@@ -155,70 +101,18 @@ decode_job(Job &job, const shared_ptr<dawn::Cmm> &cmm)
response.icc.assign(reinterpret_cast<const byte *>(image->icc.data()),
reinterpret_cast<const byte *>(image->icc.data() + image->icc.size()));
response.profile_assumed = image->profile_assumed;
response.pixmap.pixels.value = proto::BlobInline{};
proto::Frame measurement;
measurement.payload.value = proto::PayloadResponse{proto::Response{
job.request, proto::Result{proto::ResultDecoded{response}}}};
const size_t inline_overhead = encode_frame(measurement).size();
if (inline_overhead <= dawn::ipc::Connection::kMaxPayload &&
image->data.size() <=
dawn::ipc::Connection::kMaxPayload - inline_overhead) {
proto::BlobInline pixels;
pixels.bytes.assign(reinterpret_cast<const byte *>(image->data.data()),
reinterpret_cast<const byte *>(
image->data.data() + image->data.size()));
response.pixmap.pixels.value = std::move(pixels);
} else {
done.memory = dawn::ipc::SharedMemory::copy(
image->data.data(), image->data.size());
if (!done.memory.ok()) {
done.frame =
error_frame(job.request, "shared memory creation failed",
proto::ErrorCode::InvalidArgument);
return done;
}
response.pixmap.pixels.value = proto::BlobShared{image->data.size()};
}
done.frame.payload.value = proto::PayloadResponse{proto::Response{
job.request, proto::Result{proto::ResultDecoded{std::move(response)}}}};
if (done.memory.ok() &&
encode_frame(done.frame).size() > dawn::ipc::Connection::kMaxPayload) {
done.memory.close();
done.frame = error_frame(job.request, "response metadata is too large",
proto::ErrorCode::Internal);
}
return done;
}
static void
worker(State *state)
{
auto cmm = make_shared<dawn::Cmm>();
while (true) {
Job job;
{
unique_lock lock(state->mu);
state->cv.wait(
lock, [&] { return state->stop || !state->jobs.empty(); });
if (state->stop && state->jobs.empty())
return;
job = std::move(state->jobs.front());
state->jobs.pop_front();
}
Done done = decode_job(job, cmm);
const uint64_t connection = job.connection;
const size_t retained_bytes = job.retained_bytes;
job = {};
release(*state, connection, retained_bytes);
{
lock_guard lock(state->mu);
state->done.push_back(std::move(done));
}
state->loop.wake();
}
const auto encode_response = [&] {
proto::Frame frame;
frame.payload.value = proto::PayloadResponse{proto::Response{
job.request, proto::Result{proto::ResultDecoded{response}}}};
return dawn::ipc::encoded(frame);
};
if (const char *failure = dawn::ipc::place_blob(response.pixmap.pixels,
as_bytes(span(image->data)), dawn::ipc::Connection::kMaxPayload,
encode_response, reply.payload, reply.attachment))
reply.payload =
error_frame(job.request, failure, dawn::ipc::ErrorCode::Internal);
return reply;
}
// --- Server ------------------------------------------------------------------
@@ -237,21 +131,15 @@ handle_payload(State &state, uint64_t connection, span<const byte> payload,
if (!owned.empty() || state.greeted.contains(connection))
return false;
proto::DaemonHelloReply reply;
if (hello->hello.protocol_version != proto::kImagedProtocolVersion)
reply.value = proto::DaemonHelloReplyVersionMismatch{
proto::kImagedProtocolVersion};
else if (!hello->hello.session.empty())
reply.value = proto::DaemonHelloReplySessionMismatch{};
else {
reply.value = proto::DaemonHelloReplyAccepted{proto::DaemonLimits{
dawn::ipc::Connection::kMaxPayload, proto::kImagedMaxBlobSize}};
const auto reply = dawn::ipc::daemon_hello(hello->hello,
proto::kImagedProtocolVersion, proto::kImagedMaxBlobSize);
if (holds_alternative<dawn::ipc::DaemonHelloReplyAccepted>(reply.value))
state.greeted.insert(connection);
}
proto::Frame out;
out.payload.value = proto::PayloadHelloReply{reply};
return state.server->send(connection, encode_frame(out), {});
return state.host->server().send(
connection, dawn::ipc::encoded(out), {});
}
if (!state.greeted.contains(connection))
return false;
@@ -261,86 +149,37 @@ handle_payload(State &state, uint64_t connection, span<const byte> payload,
return false;
const auto invalid = [&](string message) {
return state.server->send(connection,
encode_frame(error_frame(request->request.id, std::move(message),
proto::ErrorCode::InvalidArgument)),
return state.host->server().send(connection,
error_frame(request->request.id, std::move(message),
dawn::ipc::ErrorCode::InvalidArgument),
{});
};
const auto busy = [&] {
return state.server->send(connection,
encode_frame(error_frame(request->request.id,
"image service is busy", proto::ErrorCode::Busy)),
return state.host->server().send(connection,
error_frame(request->request.id, "image service is busy",
dawn::ipc::ErrorCode::Busy),
{});
};
Job job;
job.connection = connection;
job.request = request->request.id;
const size_t icc_size = request->request.decode.target_icc.size();
auto in =
get_if<proto::BlobInlineView>(&request->request.decode.data.value);
uint64_t input_size;
if (in) {
if (!owned.empty())
return invalid("inline blob has attachments");
if (in->bytes.empty() || in->bytes.size() > kMaxBlob)
return invalid("invalid input blob size");
input_size = in->bytes.size();
} else {
auto shared =
get<proto::BlobSharedView>(request->request.decode.data.value);
if (owned.size() != 1 || !shared.size || shared.size > kMaxBlob)
return invalid("invalid shared input blob");
input_size = shared.size;
}
if (!admit(state, job, input_size + icc_size))
const auto &request_decode = request->request.decode;
auto job = make_shared<Job>();
job->request = request->request.id;
span<const byte> data;
if (const char *failure = dawn::ipc::take_blob(
request_decode.data, owned, kMaxBlob, data, job->shared))
return invalid(failure);
if (!job->shared.ok())
job->data.assign(data.begin(), data.end());
job->target_icc.assign(
request_decode.target_icc.begin(), request_decode.target_icc.end());
job->first_frame_only = request_decode.first_frame_only;
if (!state.host->submit(connection, data.size() + job->target_icc.size(),
[job] { return decode_job(*job); }))
return busy();
if (in)
job.data.assign(in->bytes.begin(), in->bytes.end());
else {
job.shared = dawn::ipc::SharedMemory::map(owned.take(0), input_size);
if (!job.shared.ok()) {
release(state, job.connection, job.retained_bytes);
return invalid("cannot map shared input blob");
}
}
job.target_icc.assign(request->request.decode.target_icc.begin(),
request->request.decode.target_icc.end());
job.first_frame_only = request->request.decode.first_frame_only;
{
lock_guard lock(state.mu);
state.jobs.push_back(std::move(job));
}
state.cv.notify_one();
return true;
}
static void
handle_read(State &state, uint64_t id)
{
if (id == dawn::ipc::Loop::kListener)
state.server->poll_listen();
else
state.server->poll_read(id);
}
static void
handle_wake(State &state)
{
deque<Done> done;
{
lock_guard lock(state.mu);
swap(done, state.done);
}
for (auto &item : done) {
auto bytes = encode_frame(item.frame);
dawn::ipc::Handle handle = item.memory.handle();
state.server->send(item.connection, bytes,
item.memory.ok() ? span(&handle, 1)
: span<const dawn::ipc::Handle>());
}
}
int
main()
{
@@ -354,6 +193,7 @@ main()
}
State state;
dawn::ipc::ServerCore::Config config;
config.on_payload = [&](uint64_t connection, span<const byte> payload,
vector<dawn::ipc::Handle> attachments) {
@@ -361,36 +201,9 @@ main()
state, connection, payload, std::move(attachments));
};
config.on_closed = [&](uint64_t id) { state.greeted.erase(id); };
config.watch_read = [&](uint64_t id, dawn::ipc::Waitable w) {
return state.loop.watch_read(id, w);
};
config.watch_write = [&](uint64_t id, dawn::ipc::Waitable w, bool on) {
state.loop.watch_write(id, w, on);
};
config.unwatch = [&](uint64_t id) { state.loop.unwatch(id); };
dawn::ipc::ServerCore server(
std::move(endpoint.listener), std::move(config));
state.server = &server;
state.loop.on_read = [&](uint64_t id) { handle_read(state, id); };
state.loop.on_write = [&](uint64_t id) { server.poll_write(id); };
state.loop.on_wake = [&] { handle_wake(state); };
if (!state.loop.watch_read(
dawn::ipc::Loop::kListener, server.listen_waitable()))
return 1;
const unsigned worker_count =
min(4u, max(1u, thread::hardware_concurrency()));
vector<thread> workers(worker_count);
for (auto &worker_thread : workers)
worker_thread = thread(worker, &state);
state.loop.run();
{
lock_guard lock(state.mu);
state.stop = true;
}
state.cv.notify_all();
for (auto &worker_thread : workers)
worker_thread.join();
dawn::ipc::DaemonHost host(std::move(endpoint.listener), std::move(config),
min(4u, max(1u, thread::hardware_concurrency())));
state.host = &host;
return host.run() ? 0 : 1;
}
dnthumbd/main.cpp
+72 -249
View File
@@ -5,114 +5,59 @@
// SPDX-License-Identifier: MPL-2.0
//
#include "ipc-loop.hpp"
#include "ipc-rpc.hpp"
#include "ipc-shm.hpp"
#include "ipc/thumbd.lxdr.hpp"
#include "libdn.h"
#include "libdnvk.h"
#include <condition_variable>
#include <cstdio>
#include <deque>
#include <mutex>
#include <thread>
#include <memory>
#include <unordered_set>
#include <variant>
#include <vector>
using namespace std;
namespace proto = dawn::ipc::thumbd;
// --- Jobs --------------------------------------------------------------------
static constexpr char kService[] = "thumbd";
static constexpr uint64_t kMaxBlob = 1024ull * 1024 * 1024;
// One scaler worker is intentionally fed from a short queue. Two jobs per
// client prevent one connection monopolizing the retained-input cap. The byte
// cap admits one advertised maximum-size request, never several of them.
static constexpr size_t kMaxJobs = 8, kMaxJobsPerConnection = 2;
static constexpr size_t kMaxRetainedBytes = 1024ull * 1024 * 1024;
static vector<byte>
encode_frame(const proto::Frame &frame)
{
vector<byte> out;
dawn::ipc::Encoder encoder(out);
encode(frame, encoder);
return out;
}
static proto::Frame
error_frame(uint64_t id, string message, proto::ErrorCode code)
error_frame(uint64_t id, string message, dawn::ipc::ErrorCode code)
{
proto::Frame frame;
frame.payload.value = proto::PayloadResponse{proto::Response{id,
proto::Result{
proto::ResultError{proto::Error{code, std::move(message)}}}}};
return frame;
proto::ResultError{dawn::ipc::Error{code, std::move(message)}}}}};
return dawn::ipc::encoded(frame);
}
namespace
{
struct Job {
uint64_t connection = 0, request = 0;
size_t retained_bytes = 0;
uint64_t request = 0;
uint32_t width = 0, height = 0, stride = 0, out_width = 0, out_height = 0;
int32_t orientation = 0;
vector<byte> inline_pixels;
dawn::ipc::SharedMemory shared;
};
struct Done {
uint64_t connection = 0;
proto::Frame frame;
dawn::ipc::SharedMemory shared;
};
struct State {
mutex mu;
condition_variable cv;
deque<Job> jobs;
deque<Done> done;
unordered_multiset<uint64_t> admitted;
unordered_set<uint64_t> greeted;
size_t retained_bytes = 0;
bool stop = false;
dawn::ipc::Loop loop;
dawn::ipc::ServerCore *server = nullptr;
dawn::ipc::DaemonHost *host = nullptr;
dawn::ScaleScaler *scaler = nullptr;
};
} // namespace
static bool
admit(State &state, Job &job, size_t bytes)
static dawn::ipc::DaemonHost::Reply
scale_job(const Job &job, dawn::ScaleScaler *scaler)
{
lock_guard lock(state.mu);
if (state.admitted.size() >= kMaxJobs ||
state.admitted.count(job.connection) >= kMaxJobsPerConnection ||
bytes > kMaxRetainedBytes ||
state.retained_bytes > kMaxRetainedBytes - bytes)
return false;
state.admitted.insert(job.connection);
state.retained_bytes += bytes;
job.retained_bytes = bytes;
return true;
}
static void
release(State &state, uint64_t connection, size_t bytes)
{
lock_guard lock(state.mu);
state.retained_bytes -= bytes;
state.admitted.erase(state.admitted.find(connection));
}
static Done
scale_job(Job &job, dawn::ScaleScaler *scaler)
{
Done done;
done.connection = job.connection;
dawn::ipc::DaemonHost::Reply reply;
dawn::ScaleOutput output;
string error;
const uint8_t *pixels = job.shared.ok()
@@ -122,78 +67,27 @@ scale_job(Job &job, dawn::ScaleScaler *scaler)
job.out_height,
dawn::orientation_or_0(dawn::Orientation(job.orientation)), &output,
&error)) {
done.frame =
reply.payload =
error_frame(job.request, error.empty() ? "scale failed" : error,
proto::ErrorCode::InvalidArgument);
return done;
dawn::ipc::ErrorCode::InvalidArgument);
return reply;
}
proto::ScaleResponse response;
response.width = output.width;
response.height = output.height;
response.rgba8.value = proto::BlobInline{};
proto::Frame measurement;
measurement.payload.value = proto::PayloadResponse{proto::Response{
job.request, proto::Result{proto::ResultScaled{response}}}};
const size_t inline_overhead = encode_frame(measurement).size();
if (inline_overhead <= dawn::ipc::Connection::kMaxPayload &&
output.rgba8.size() <=
dawn::ipc::Connection::kMaxPayload - inline_overhead) {
proto::BlobInline blob;
blob.bytes.assign(reinterpret_cast<const byte *>(output.rgba8.data()),
reinterpret_cast<const byte *>(
output.rgba8.data() + output.rgba8.size()));
response.rgba8.value = std::move(blob);
} else {
done.shared = dawn::ipc::SharedMemory::copy(
output.rgba8.data(), output.rgba8.size());
if (!done.shared.ok()) {
done.frame =
error_frame(job.request, "shared memory creation failed",
proto::ErrorCode::InvalidArgument);
return done;
}
response.rgba8.value = proto::BlobShared{output.rgba8.size()};
}
done.frame.payload.value = proto::PayloadResponse{proto::Response{
job.request, proto::Result{proto::ResultScaled{std::move(response)}}}};
if (done.shared.ok() &&
encode_frame(done.frame).size() > dawn::ipc::Connection::kMaxPayload) {
done.shared.close();
done.frame = error_frame(job.request, "response metadata is too large",
proto::ErrorCode::Internal);
}
return done;
}
static void
worker(State *state, dawn::ScaleScaler *scaler)
{
while (true) {
Job job;
{
unique_lock lock(state->mu);
state->cv.wait(
lock, [&] { return state->stop || !state->jobs.empty(); });
if (state->stop && state->jobs.empty())
return;
job = std::move(state->jobs.front());
state->jobs.pop_front();
}
Done done = scale_job(job, scaler);
const uint64_t connection = job.connection;
const size_t retained_bytes = job.retained_bytes;
job = {};
release(*state, connection, retained_bytes);
{
lock_guard lock(state->mu);
state->done.push_back(std::move(done));
}
state->loop.wake();
}
const auto encode_response = [&] {
proto::Frame frame;
frame.payload.value = proto::PayloadResponse{proto::Response{
job.request, proto::Result{proto::ResultScaled{response}}}};
return dawn::ipc::encoded(frame);
};
if (const char *failure = dawn::ipc::place_blob(response.rgba8,
as_bytes(span(output.rgba8)), dawn::ipc::Connection::kMaxPayload,
encode_response, reply.payload, reply.attachment))
reply.payload =
error_frame(job.request, failure, dawn::ipc::ErrorCode::Internal);
return reply;
}
// --- Server ------------------------------------------------------------------
@@ -212,21 +106,15 @@ handle_payload(State &state, uint64_t connection, span<const byte> payload,
if (!owned.empty() || state.greeted.contains(connection))
return false;
proto::DaemonHelloReply reply;
if (hello->hello.protocol_version != proto::kThumbdProtocolVersion)
reply.value = proto::DaemonHelloReplyVersionMismatch{
proto::kThumbdProtocolVersion};
else if (!hello->hello.session.empty())
reply.value = proto::DaemonHelloReplySessionMismatch{};
else {
reply.value = proto::DaemonHelloReplyAccepted{proto::DaemonLimits{
dawn::ipc::Connection::kMaxPayload, proto::kThumbdMaxBlobSize}};
const auto reply = dawn::ipc::daemon_hello(hello->hello,
proto::kThumbdProtocolVersion, proto::kThumbdMaxBlobSize);
if (holds_alternative<dawn::ipc::DaemonHelloReplyAccepted>(reply.value))
state.greeted.insert(connection);
}
proto::Frame out;
out.payload.value = proto::PayloadHelloReply{reply};
return state.server->send(connection, encode_frame(out), {});
return state.host->server().send(
connection, dawn::ipc::encoded(out), {});
}
if (!state.greeted.contains(connection))
return false;
@@ -236,102 +124,55 @@ handle_payload(State &state, uint64_t connection, span<const byte> payload,
return false;
const auto invalid = [&](string message) {
return state.server->send(connection,
encode_frame(error_frame(request->request.id, std::move(message),
proto::ErrorCode::InvalidArgument)),
return state.host->server().send(connection,
error_frame(request->request.id, std::move(message),
dawn::ipc::ErrorCode::InvalidArgument),
{});
};
const auto busy = [&] {
return state.server->send(connection,
encode_frame(error_frame(request->request.id,
"thumbnail service is busy", proto::ErrorCode::Busy)),
return state.host->server().send(connection,
error_frame(request->request.id, "thumbnail service is busy",
dawn::ipc::ErrorCode::Busy),
{});
};
const auto &scale = request->request.scale;
Job job;
job.connection = connection;
job.request = request->request.id;
job.width = scale.source.width;
job.height = scale.source.height;
job.stride = scale.source.stride;
job.orientation = scale.source.orientation;
job.out_width = scale.out_width;
job.out_height = scale.out_height;
const uint64_t row = uint64_t(job.width) * dawn::kBytesPerPixel;
if (job.orientation < 0 || job.orientation > 8)
auto job = make_shared<Job>();
job->request = request->request.id;
job->width = scale.source.width;
job->height = scale.source.height;
job->stride = scale.source.stride;
job->orientation = scale.source.orientation;
job->out_width = scale.out_width;
job->out_height = scale.out_height;
const uint64_t row = uint64_t(job->width) * dawn::kBytesPerPixel;
if (job->orientation < 0 || job->orientation > 8)
return invalid("invalid image orientation");
if (!job.width || !job.height || job.width > dawn::kMaxDimension ||
job.height > dawn::kMaxDimension || !job.out_width || !job.out_height ||
job.out_width > dawn::kMaxDimension ||
job.out_height > dawn::kMaxDimension || job.stride < row ||
uint64_t(job.out_width) * job.out_height * 4 > kMaxBlob)
if (!job->width || !job->height || job->width > dawn::kMaxDimension ||
job->height > dawn::kMaxDimension || !job->out_width ||
!job->out_height || job->out_width > dawn::kMaxDimension ||
job->out_height > dawn::kMaxDimension || job->stride < row ||
uint64_t(job->out_width) * job->out_height * 4 > kMaxBlob)
return invalid("invalid image geometry");
auto in = get_if<proto::BlobInlineView>(&scale.source.pixels.value);
uint64_t input_size;
if (in) {
if (!owned.empty() ||
uint64_t(job.stride) * job.height > in->bytes.size())
return invalid("invalid inline pixel blob");
if (in->bytes.size() > kMaxBlob)
return invalid("pixel blob is too large");
input_size = in->bytes.size();
} else {
auto shared = get<proto::BlobSharedView>(scale.source.pixels.value);
if (owned.size() != 1 || !shared.size || shared.size > kMaxBlob ||
uint64_t(job.stride) * job.height > shared.size)
return invalid("invalid shared pixel blob");
input_size = shared.size;
}
if (!admit(state, job, input_size))
span<const byte> pixels;
if (const char *failure = dawn::ipc::take_blob(
scale.source.pixels, owned, kMaxBlob, pixels, job->shared))
return invalid(failure);
if (uint64_t(job->stride) * job->height > pixels.size())
return invalid("pixel blob is too small");
if (!job->shared.ok())
job->inline_pixels.assign(pixels.begin(), pixels.end());
if (!state.host->submit(connection, pixels.size(),
[job, scaler = state.scaler] { return scale_job(*job, scaler); }))
return busy();
if (in)
job.inline_pixels.assign(in->bytes.begin(), in->bytes.end());
else {
job.shared = dawn::ipc::SharedMemory::map(owned.take(0), input_size);
if (!job.shared.ok()) {
release(state, job.connection, job.retained_bytes);
return invalid("cannot map shared pixel blob");
}
}
{
lock_guard lock(state.mu);
state.jobs.push_back(std::move(job));
}
state.cv.notify_one();
return true;
}
static void
handle_read(State &state, uint64_t id)
{
if (id == dawn::ipc::Loop::kListener)
state.server->poll_listen();
else
state.server->poll_read(id);
}
static void
handle_wake(State &state)
{
deque<Done> done;
{
lock_guard lock(state.mu);
swap(done, state.done);
}
for (auto &item : done) {
auto bytes = encode_frame(item.frame);
dawn::ipc::Handle handle = item.shared.handle();
state.server->send(item.connection, bytes,
item.shared.ok() ? span(&handle, 1)
: span<const dawn::ipc::Handle>());
}
}
int
main()
{
auto endpoint = dawn::ipc::Endpoint::listen("thumbd");
auto endpoint = dawn::ipc::Endpoint::listen(kService);
if (endpoint.status != dawn::ipc::Endpoint::ListenStatus::Ok) {
fprintf(stderr, "dnthumbd: %s\n",
endpoint.status == dawn::ipc::Endpoint::ListenStatus::InUse
@@ -348,6 +189,8 @@ main()
}
State state;
state.scaler = &scaler;
dawn::ipc::ServerCore::Config config;
config.on_payload = [&](uint64_t connection, span<const byte> payload,
vector<dawn::ipc::Handle> attachments) {
@@ -355,30 +198,10 @@ main()
state, connection, payload, std::move(attachments));
};
config.on_closed = [&](uint64_t id) { state.greeted.erase(id); };
config.watch_read = [&](uint64_t id, dawn::ipc::Waitable w) {
return state.loop.watch_read(id, w);
};
config.watch_write = [&](uint64_t id, dawn::ipc::Waitable w, bool on) {
state.loop.watch_write(id, w, on);
};
config.unwatch = [&](uint64_t id) { state.loop.unwatch(id); };
dawn::ipc::ServerCore server(
std::move(endpoint.listener), std::move(config));
state.server = &server;
state.loop.on_read = [&](uint64_t id) { handle_read(state, id); };
state.loop.on_write = [&](uint64_t id) { server.poll_write(id); };
state.loop.on_wake = [&] { handle_wake(state); };
if (!state.loop.watch_read(
dawn::ipc::Loop::kListener, server.listen_waitable()))
return 1;
thread thread(worker, &state, &scaler);
state.loop.run();
{
lock_guard lock(state.mu);
state.stop = true;
}
state.cv.notify_all();
thread.join();
// The scaler serializes internally, so more workers would only queue.
dawn::ipc::DaemonHost host(
std::move(endpoint.listener), std::move(config), 1);
state.host = &host;
return host.run() ? 0 : 1;
}
ipc/lxdrgen-cpp.awk
+45 -6
View File
@@ -5,10 +5,15 @@
#
# Usage: env LC_ALL=C awk -f lxdrgen.awk -f lxdrgen-cpp.awk \
# -v PrefixCamel=Instance -v Namespace=dawn::ipc::instance \
# common.lxdr instance.lxdr
# -v Extern=ipc/common.lxdr common.lxdr instance.lxdr
#
# PrefixCamel is accepted for consistency with other backends and is not
# applied to C++ type names. IDL names are emitted as-is.
#
# Extern names .lxdr files, by their include stem, that are read only to
# learn their types: nothing is emitted for them, and the header includes
# <stem>.hpp instead. They have to come first, and their namespace has to
# enclose Namespace, so that the types resolve unqualified.
function define_internal(name, ctype, view) {
Types[name] = "internal"
@@ -126,6 +131,11 @@ function emit_preamble( i, src) {
print "#include <string_view>"
print "#include <variant>"
print "#include <vector>"
for (i = 1; i <= ExternCount; i++) {
if (i == 1)
print ""
print "#include \"" ExternList[i] ".hpp\""
}
print ""
print "namespace dawn::ipc {"
print "class Encoder;"
@@ -138,6 +148,17 @@ function emit_preamble( i, src) {
BEGIN {
if (Namespace == "")
Namespace = "dawn::ipc"
ExternCount = split(Extern, ExternList, " ")
for (ExternIndex = 1; ExternIndex <= ExternCount; ExternIndex++)
ExternFiles[basename(ExternList[ExternIndex])] = 1
}
# Extern files are read first, so the switch is one-way: the first file
# that is not one of them turns emission on for the rest of the input.
function emitting() {
if (!Emitting && !(basename(FILENAME) in ExternFiles))
Emitting = 1
return Emitting
}
function codegen_begin() {
@@ -169,6 +190,9 @@ function codegen_begin() {
}
function codegen_constant(name, value) {
if (!emitting())
return
print ""
print "inline constexpr int32_t k" snaketocamel(name) " = " value ";"
}
@@ -180,7 +204,7 @@ function codegen_enum_value(name, subname, value, cg) {
"\tcase " name "::" snaketocamel(subname) ":\n")
}
function codegen_enum(name, cg, fields) {
function emit_enum(name, cg, fields) {
fields = cg["fields"]
sub(/,\n$/, "\n", fields)
@@ -211,6 +235,11 @@ function codegen_enum(name, cg, fields) {
print "\t\treturn false;"
print "\t}"
print "}"
}
function codegen_enum(name, cg, i) {
if (emitting())
emit_enum(name, cg)
CodegenSerialize[name] = "\tencode(%s, encoder);\n"
CodegenDeserialize[name] = \
@@ -281,7 +310,7 @@ function emit_codec(own, view, ser, des) {
}
}
function codegen_struct(name, cg, view) {
function emit_struct(name, cg, view) {
view = name "View"
print ""
@@ -295,13 +324,18 @@ function codegen_struct(name, cg, view) {
print "};"
emit_codec(name, view, cg["serialize"], cg["deserialize"])
}
function codegen_struct(name, cg, i) {
if (emitting())
emit_struct(name, cg)
CodegenSerialize[name] = "\tencode(%s, encoder);\n"
CodegenDeserialize[name] = \
"\tif (!decode(decoder, %s))\n" \
"\t\treturn false;\n"
CodegenCType[name] = name
CodegenViewType[name] = view
CodegenViewType[name] = name "View"
for (i in cg)
delete cg[i]
}
@@ -322,7 +356,7 @@ function codegen_union_struct(name, casename, cg, scg, n, arm) {
cg["case", n] = snaketocamel(casename)
}
function codegen_union(name, cg, view, n, i, own_list, view_list, tag) {
function emit_union(name, cg, view, n, i, own_list, view_list, tag) {
view = name "View"
n = cg["n"] + 0
tag = cg["tagtype"]
@@ -383,13 +417,18 @@ function codegen_union(name, cg, view, n, i, own_list, view_list, tag) {
print "\t\treturn false;"
print "\t}"
print "}"
}
function codegen_union(name, cg, i) {
if (emitting())
emit_union(name, cg)
CodegenSerialize[name] = "\tencode(%s, encoder);\n"
CodegenDeserialize[name] = \
"\tif (!decode(decoder, %s))\n" \
"\t\treturn false;\n"
CodegenCType[name] = name
CodegenViewType[name] = view
CodegenViewType[name] = name "View"
for (i in cg)
delete cg[i]
}
libdn/CMakeLists.txt
+13 -5
View File
@@ -133,33 +133,41 @@ elseif (NOT APPLE)
)
endif()
# Generated LXDR header; ipc-instance.cpp binds it to the RPC core, and
# tests/test-lxdr.cpp checks it against Encoder/Decoder.
# Generated LXDR headers; ipc-instance.cpp binds instance to the RPC core,
# and tests/test-lxdr.cpp checks them against Encoder/Decoder. The types
# every service shares are generated once, into the enclosing namespace.
lxdr_generate(
${PROJECT_BINARY_DIR}/generated/ipc/common.lxdr.hpp
dawn::ipc
Common
${PROJECT_SOURCE_DIR}/ipc/common.lxdr
)
lxdr_generate(
${PROJECT_BINARY_DIR}/generated/ipc/instance.lxdr.hpp
dawn::ipc::instance
Instance
${PROJECT_SOURCE_DIR}/ipc/common.lxdr
${PROJECT_SOURCE_DIR}/ipc/instance.lxdr
EXTERN ${PROJECT_SOURCE_DIR}/ipc/common.lxdr
)
lxdr_generate(
${PROJECT_BINARY_DIR}/generated/ipc/imaged.lxdr.hpp
dawn::ipc::imaged
Imaged
${PROJECT_SOURCE_DIR}/ipc/common.lxdr
${PROJECT_SOURCE_DIR}/ipc/imaged.lxdr
EXTERN ${PROJECT_SOURCE_DIR}/ipc/common.lxdr
)
lxdr_generate(
${PROJECT_BINARY_DIR}/generated/ipc/thumbd.lxdr.hpp
dawn::ipc::thumbd
Thumbd
${PROJECT_SOURCE_DIR}/ipc/common.lxdr
${PROJECT_SOURCE_DIR}/ipc/thumbd.lxdr
EXTERN ${PROJECT_SOURCE_DIR}/ipc/common.lxdr
)
add_custom_target(dn-tiff-tables
DEPENDS ${CMAKE_CURRENT_BINARY_DIR}/tiff-tables.h)
add_dependencies(libdn
dn-tiff-tables
lxdr-common_lxdr_hpp
lxdr-imaged_lxdr_hpp
lxdr-instance_lxdr_hpp
lxdr-thumbd_lxdr_hpp
libdn/ipc-rpc.cpp
+178
View File
@@ -9,7 +9,9 @@
#include <climits>
#include <thread>
#include <utility>
#include <variant>
using namespace std;
@@ -314,5 +316,181 @@ ServerCore::drop(uint64_t id)
cfg_.on_closed(id);
}
// --- Daemons -----------------------------------------------------------------
const char *
place_blob(Blob &blob, span<const byte> bytes, uint32_t limit,
const function<vector<byte>()> &encode, vector<byte> &out,
SharedMemory &attachment)
{
blob.value = BlobInline{};
const size_t overhead = encode().size();
if (overhead <= limit && bytes.size() <= limit - overhead) {
blob.value = BlobInline{vector<byte>(bytes.begin(), bytes.end())};
out = encode();
return nullptr;
}
attachment = SharedMemory::copy(bytes.data(), bytes.size());
if (!attachment.ok())
return "shared memory creation failed";
blob.value = BlobShared{bytes.size()};
out = encode();
if (out.size() <= limit)
return nullptr;
attachment.close();
return "frame metadata is too large";
}
const char *
take_blob(const BlobView &blob, OwnedHandles &handles, uint64_t limit,
span<const byte> &bytes, SharedMemory &memory)
{
if (auto in = get_if<BlobInlineView>(&blob.value)) {
if (!handles.empty())
return "inline blob has attachments";
if (in->bytes.empty() || in->bytes.size() > limit)
return "invalid inline blob size";
bytes = in->bytes;
return nullptr;
}
const auto shared = get<BlobSharedView>(blob.value);
if (handles.size() != 1 || !shared.size || shared.size > limit)
return "invalid shared blob";
memory = SharedMemory::map(handles.take(0), shared.size);
if (!memory.ok())
return "cannot map shared blob";
bytes = {reinterpret_cast<const byte *>(memory.data()), memory.size()};
return nullptr;
}
DaemonHelloReply
daemon_hello(const HelloView &hello, uint32_t version, uint64_t max_blob_size)
{
DaemonHelloReply reply;
if (hello.protocol_version != version)
reply.value = DaemonHelloReplyVersionMismatch{version};
else if (!hello.session.empty())
reply.value = DaemonHelloReplySessionMismatch{};
else
reply.value = DaemonHelloReplyAccepted{
DaemonLimits{Connection::kMaxPayload, max_blob_size}};
return reply;
}
DaemonHost::DaemonHost(
Listener listener, ServerCore::Config cfg, unsigned workers)
: workers_(workers)
{
cfg.watch_read = [this](uint64_t id, Waitable w) {
return loop_.watch_read(id, w);
};
cfg.unwatch = [this](uint64_t id) { loop_.unwatch(id); };
cfg.watch_write = [this](uint64_t id, Waitable w, bool enable) {
loop_.watch_write(id, w, enable);
};
core_ = make_unique<ServerCore>(std::move(listener), std::move(cfg));
loop_.on_read = [this](uint64_t id) {
if (id == Loop::kListener)
core_->poll_listen();
else
core_->poll_read(id);
};
loop_.on_write = [this](uint64_t id) { core_->poll_write(id); };
loop_.on_wake = [this] { this->flush_done(); };
}
bool
DaemonHost::submit(uint64_t id, size_t bytes, Work work)
{
{
lock_guard lock(mutex_);
if (admitted_.size() >= kMaxJobs ||
admitted_.count(id) >= kMaxJobsPerConnection ||
bytes > kMaxRetainedBytes ||
retained_bytes_ > kMaxRetainedBytes - bytes)
return false;
admitted_.insert(id);
retained_bytes_ += bytes;
jobs_.push_back(Job{id, bytes, std::move(work)});
}
cv_.notify_one();
return true;
}
void
DaemonHost::work()
{
while (true) {
Job job;
{
unique_lock lock(mutex_);
cv_.wait(lock, [&] { return stop_ || !jobs_.empty(); });
if (stop_ && jobs_.empty())
return;
job = std::move(jobs_.front());
jobs_.pop_front();
}
Reply reply = job.work();
// Dropped before the accounting, because whatever input the work
// was holding on to goes with it.
const uint64_t id = job.connection;
const size_t bytes = job.bytes;
job = {};
{
lock_guard lock(mutex_);
retained_bytes_ -= bytes;
admitted_.erase(admitted_.find(id));
done_.emplace_back(id, std::move(reply));
}
loop_.wake();
}
}
void
DaemonHost::flush_done()
{
deque<pair<uint64_t, Reply>> done;
{
lock_guard lock(mutex_);
swap(done, done_);
}
for (auto &[id, reply] : done) {
const Handle handle = reply.attachment.handle();
core_->send(id, reply.payload,
reply.attachment.ok() ? span(&handle, 1) : span<const Handle>());
}
}
bool
DaemonHost::run()
{
if (!loop_.watch_read(Loop::kListener, core_->listen_waitable()))
return false;
vector<thread> workers;
for (unsigned i = 0; i < workers_; i++)
workers.emplace_back([this] { this->work(); });
loop_.run();
{
lock_guard lock(mutex_);
stop_ = true;
}
cv_.notify_all();
for (auto &worker : workers)
worker.join();
return true;
}
} // namespace ipc
} // namespace dawn
libdn/ipc-rpc.hpp
+97 -2
View File
@@ -7,14 +7,23 @@
#pragma once
#include "ipc-loop.hpp"
#include "ipc-shm.hpp"
#include "ipc.hpp"
#include "ipc/common.lxdr.hpp"
#include <chrono>
#include <condition_variable>
#include <cstdint>
#include <deque>
#include <functional>
#include <memory>
#include <mutex>
#include <span>
#include <string_view>
#include <unordered_map>
#include <unordered_set>
#include <utility>
#include <vector>
namespace dawn
@@ -22,8 +31,9 @@ namespace dawn
namespace ipc
{
// Nothing here knows any schema: payloads go in and out whole. A service
// puts its own handshake and envelope dispatch on top.
// Nothing here knows any service's schema: payloads go in and out whole,
// and only the types every service shares are understood. A service puts
// its own handshake and envelope dispatch on top.
// Timeouts are asymmetric: a handshake that does not complete promptly means
// running a second process, while a request already sent must be given time
@@ -130,5 +140,90 @@ private:
std::unordered_map<uint64_t, Conn> conns_;
};
// --- Daemons -----------------------------------------------------------------
// Hand bytes to a blob: inline while the frame that encode() renders
// around it stays within limit, in shared memory otherwise. Leaves the
// encoded frame in out, and the memory in attachment when it took that
// route. Returns what went wrong, or null.
const char *place_blob(Blob &blob, std::span<const std::byte> bytes,
uint32_t limit, const std::function<std::vector<std::byte>()> &encode,
std::vector<std::byte> &out, SharedMemory &attachment);
// Where a blob's bytes are, mapping the frame's one attachment into
// memory when that is the route they took. Bounds the size by limit,
// but says nothing about what the service considers a valid one.
// Returns what went wrong, or null.
const char *take_blob(const BlobView &blob, OwnedHandles &handles,
uint64_t limit, std::span<const std::byte> &bytes, SharedMemory &memory);
// The one answer a daemon gives to a Hello. Anything but Accepted means
// the peer must not go on to make requests.
DaemonHelloReply daemon_hello(
const HelloView &hello, uint32_t version, uint64_t max_blob_size);
// A daemon: a ServerCore, the event loop that drives it, and worker
// threads fed from a bounded queue. It knows no service's schema -- the
// service decodes a request, submits the work as a closure, and gets to
// encode whatever comes back.
class DaemonHost
{
public:
// One finished unit of work: the response payload, and the shared
// memory it points at, which rides along as its one attachment.
struct Reply {
std::vector<std::byte> payload;
SharedMemory attachment;
};
// Runs on a worker thread. It has to be copyable, which is what
// keeping the job itself in a shared_ptr is for.
using Work = std::function<Reply()>;
// One worker is fed from a short queue. Two jobs per client prevent
// one connection monopolizing the retained-input cap, which admits
// one advertised maximum-size request, never several of them.
static constexpr size_t kMaxJobs = 8, kMaxJobsPerConnection = 2;
static constexpr size_t kMaxRetainedBytes =
1024ull * 1024 * 1024 + Connection::kMaxPayload;
// The event-loop hooks in cfg are the host's own, and are ignored.
DaemonHost(Listener listener, ServerCore::Config cfg, unsigned workers);
DaemonHost(const DaemonHost &) = delete;
DaemonHost &operator=(const DaemonHost &) = delete;
// What to answer requests with, on the loop thread.
[[nodiscard]] ServerCore &server() { return *core_; }
// Queue one unit of work, charging bytes against the retained-input
// budget. False means the daemon is busy and nothing was queued.
bool submit(uint64_t id, size_t bytes, Work work);
// Serve until the loop gives up, then drain the workers.
bool run();
private:
struct Job {
uint64_t connection = 0;
size_t bytes = 0;
Work work;
};
unsigned workers_ = 1;
Loop loop_;
std::unique_ptr<ServerCore> core_;
std::mutex mutex_;
std::condition_variable cv_;
std::deque<Job> jobs_;
std::deque<std::pair<uint64_t, Reply>> done_;
std::unordered_multiset<uint64_t> admitted_;
size_t retained_bytes_ = 0;
bool stop_ = false;
void work();
void flush_done();
};
} // namespace ipc
} // namespace dawn
libdn/ipc.hpp
+12
View File
@@ -89,6 +89,18 @@ template <typename TView> struct Received {
TView view;
};
// Any service's frame, on the wire. What encodes it is found by ADL,
// beside the frame's own type.
template <typename T>
std::vector<std::byte>
encoded(const T &frame)
{
std::vector<std::byte> out;
Encoder encoder(out);
encode(frame, encoder);
return out;
}
// A file descriptor on POSIX, a HANDLE on Windows. Both compare equal to
// -1 when invalid.
using Handle = std::intptr_t;
tests/test-ipc.cpp
+11 -11
View File
@@ -664,7 +664,7 @@ frame_bytes(const inst::Frame &frame)
static inst::Frame
hello_frame(uint32_t version, string_view session)
{
inst::Hello hello;
dawn::ipc::Hello hello;
hello.protocol_version = version;
hello.session = string(session);
inst::Frame frame;
@@ -689,7 +689,7 @@ static inst::Frame
cancel_frame(uint64_t id)
{
inst::Frame frame;
frame.payload.value = inst::PayloadCancel{inst::Cancel{id}};
frame.payload.value = inst::PayloadCancel{dawn::ipc::Cancel{id}};
return frame;
}
@@ -827,7 +827,7 @@ handshake(Fixture &f, uint32_t &limit)
if (!reply)
return false;
const auto *accepted =
get_if<inst::HelloReplyAcceptedView>(&reply->hello_reply.value);
get_if<dawn::ipc::HelloReplyAcceptedView>(&reply->hello_reply.value);
if (!accepted)
return false;
limit = accepted->limits.max_payload_size;
@@ -876,8 +876,8 @@ test_instance_version_mismatch()
CHECK(reply != nullptr);
if (!reply)
return;
const auto *mismatch =
get_if<inst::HelloReplyVersionMismatchView>(&reply->hello_reply.value);
const auto *mismatch = get_if<dawn::ipc::HelloReplyVersionMismatchView>(
&reply->hello_reply.value);
CHECK(mismatch != nullptr);
if (mismatch) {
CHECK(mismatch->server_protocol_version ==
@@ -903,7 +903,7 @@ test_instance_session_mismatch()
get_if<inst::PayloadHelloReplyView>(&view.payload.value);
CHECK(reply != nullptr);
if (reply) {
CHECK(holds_alternative<inst::HelloReplySessionMismatchView>(
CHECK(holds_alternative<dawn::ipc::HelloReplySessionMismatchView>(
reply->hello_reply.value));
}
CHECK(f.closed());
@@ -988,7 +988,7 @@ test_instance_out_of_order()
// Answered back to front; each response still carries its own ID.
held[1].done();
held[0].fail(inst::ErrorCode::NotFound, "gone");
held[0].fail(dawn::ipc::ErrorCode::NotFound, "gone");
inst::FrameView view{};
vector<byte> storage;
@@ -1014,7 +1014,7 @@ test_instance_out_of_order()
get_if<inst::ResultErrorView>(&second->response.result.value);
CHECK(err != nullptr);
if (err) {
CHECK(err->error.code == inst::ErrorCode::NotFound);
CHECK(err->error.code == dawn::ipc::ErrorCode::NotFound);
CHECK(err->error.message == "gone");
}
}
@@ -1050,7 +1050,7 @@ test_instance_cancel()
CHECK(held[0].cancelled());
// Cancellation is not itself an answer; the service still sends one.
held[0].fail(inst::ErrorCode::Cancelled, {});
held[0].fail(dawn::ipc::ErrorCode::Cancelled, {});
inst::FrameView view{};
vector<byte> storage;
CHECK(f.recv(view, storage));
@@ -1063,7 +1063,7 @@ test_instance_cancel()
get_if<inst::ResultErrorView>(&resp->response.result.value);
CHECK(err != nullptr);
if (err)
CHECK(err->error.code == inst::ErrorCode::Cancelled);
CHECK(err->error.code == dawn::ipc::ErrorCode::Cancelled);
}
static void
@@ -1109,7 +1109,7 @@ test_instance_dropped_call()
get_if<inst::ResultErrorView>(&resp->response.result.value);
CHECK(err != nullptr);
if (err)
CHECK(err->error.code == inst::ErrorCode::Internal);
CHECK(err->error.code == dawn::ipc::ErrorCode::Internal);
}
static void
tests/test-lxdr.cpp
+84 -79
View File
@@ -28,7 +28,7 @@ namespace thumbd = dawn::ipc::thumbd;
template <typename T>
static vector<byte>
encoded(const T &value)
wire(const T &value)
{
vector<byte> buf;
dawn::ipc::Encoder enc(buf);
@@ -66,40 +66,41 @@ static void
test_round_trip_hello_reply()
{
{
inst::HelloReply own;
own.value = inst::HelloReplyAccepted{inst::Limits{0x00100000u}};
const vector<byte> buf = encoded(own);
dawn::ipc::HelloReply own;
own.value =
dawn::ipc::HelloReplyAccepted{dawn::ipc::Limits{0x00100000u}};
const vector<byte> buf = wire(own);
dawn::ipc::Decoder dec(buf);
inst::HelloReplyView view{};
dawn::ipc::HelloReplyView view{};
CHECK(decode(dec, view));
CHECK(dec.remaining() == 0);
CHECK(holds_alternative<inst::HelloReplyAcceptedView>(view.value));
CHECK(get<inst::HelloReplyAcceptedView>(view.value)
CHECK(holds_alternative<dawn::ipc::HelloReplyAcceptedView>(view.value));
CHECK(get<dawn::ipc::HelloReplyAcceptedView>(view.value)
.limits.max_payload_size == 0x00100000u);
}
{
inst::HelloReply own;
own.value = inst::HelloReplyVersionMismatch{2};
const vector<byte> buf = encoded(own);
dawn::ipc::HelloReply own;
own.value = dawn::ipc::HelloReplyVersionMismatch{2};
const vector<byte> buf = wire(own);
dawn::ipc::Decoder dec(buf);
inst::HelloReplyView view{};
dawn::ipc::HelloReplyView view{};
CHECK(decode(dec, view));
CHECK(dec.remaining() == 0);
CHECK(
holds_alternative<inst::HelloReplyVersionMismatchView>(view.value));
CHECK(get<inst::HelloReplyVersionMismatchView>(view.value)
CHECK(holds_alternative<dawn::ipc::HelloReplyVersionMismatchView>(
view.value));
CHECK(get<dawn::ipc::HelloReplyVersionMismatchView>(view.value)
.server_protocol_version == 2);
}
{
inst::HelloReply own;
own.value = inst::HelloReplySessionMismatch{};
const vector<byte> buf = encoded(own);
dawn::ipc::HelloReply own;
own.value = dawn::ipc::HelloReplySessionMismatch{};
const vector<byte> buf = wire(own);
dawn::ipc::Decoder dec(buf);
inst::HelloReplyView view{};
dawn::ipc::HelloReplyView view{};
CHECK(decode(dec, view));
CHECK(dec.remaining() == 0);
CHECK(
holds_alternative<inst::HelloReplySessionMismatchView>(view.value));
CHECK(holds_alternative<dawn::ipc::HelloReplySessionMismatchView>(
view.value));
}
}
@@ -109,7 +110,7 @@ test_round_trip_result()
{
inst::Result own;
own.value = inst::ResultDone{};
const vector<byte> buf = encoded(own);
const vector<byte> buf = wire(own);
dawn::ipc::Decoder dec(buf);
inst::ResultView view{};
CHECK(decode(dec, view));
@@ -118,16 +119,16 @@ test_round_trip_result()
}
{
inst::Result own;
own.value =
inst::ResultError{inst::Error{inst::ErrorCode::Busy, "nope"}};
const vector<byte> buf = encoded(own);
own.value = inst::ResultError{
dawn::ipc::Error{dawn::ipc::ErrorCode::Busy, "nope"}};
const vector<byte> buf = wire(own);
dawn::ipc::Decoder dec(buf);
inst::ResultView view{};
CHECK(decode(dec, view));
CHECK(dec.remaining() == 0);
CHECK(holds_alternative<inst::ResultErrorView>(view.value));
const auto &err = get<inst::ResultErrorView>(view.value).error;
CHECK(err.code == inst::ErrorCode::Busy);
CHECK(err.code == dawn::ipc::ErrorCode::Busy);
CHECK(err.message == "nope");
}
}
@@ -136,10 +137,10 @@ static void
test_round_trip_frames()
{
{
inst::Hello hello{1, "sess"};
const vector<byte> buf = encoded(hello);
dawn::ipc::Hello hello{1, "sess"};
const vector<byte> buf = wire(hello);
dawn::ipc::Decoder dec(buf);
inst::HelloView view{};
dawn::ipc::HelloView view{};
CHECK(decode(dec, view));
CHECK(dec.remaining() == 0);
CHECK(view.protocol_version == 1);
@@ -150,7 +151,7 @@ test_round_trip_frames()
open.urls = {"/one", "/two"};
open.activation_token = "tok";
open.browse = true;
const vector<byte> buf = encoded(open);
const vector<byte> buf = wire(open);
dawn::ipc::Decoder dec(buf);
inst::OpenRequestView view{};
CHECK(decode(dec, view));
@@ -164,8 +165,8 @@ test_round_trip_frames()
{
inst::Frame own;
own.payload.value = inst::PayloadHello{inst::Hello{1, "b"}};
const vector<byte> buf = encoded(own);
own.payload.value = inst::PayloadHello{dawn::ipc::Hello{1, "b"}};
const vector<byte> buf = wire(own);
dawn::ipc::Decoder dec(buf);
inst::FrameView view{};
CHECK(decode(dec, view));
@@ -176,11 +177,11 @@ test_round_trip_frames()
CHECK(h.session == "b");
}
{
inst::HelloReply reply;
reply.value = inst::HelloReplyAccepted{inst::Limits{4096}};
dawn::ipc::HelloReply reply;
reply.value = dawn::ipc::HelloReplyAccepted{dawn::ipc::Limits{4096}};
inst::Frame own;
own.payload.value = inst::PayloadHelloReply{reply};
const vector<byte> buf = encoded(own);
const vector<byte> buf = wire(own);
dawn::ipc::Decoder dec(buf);
inst::FrameView view{};
CHECK(decode(dec, view));
@@ -189,8 +190,8 @@ test_round_trip_frames()
holds_alternative<inst::PayloadHelloReplyView>(view.payload.value));
const auto &hr =
get<inst::PayloadHelloReplyView>(view.payload.value).hello_reply;
CHECK(holds_alternative<inst::HelloReplyAcceptedView>(hr.value));
CHECK(get<inst::HelloReplyAcceptedView>(hr.value)
CHECK(holds_alternative<dawn::ipc::HelloReplyAcceptedView>(hr.value));
CHECK(get<dawn::ipc::HelloReplyAcceptedView>(hr.value)
.limits.max_payload_size == 4096);
}
{
@@ -202,7 +203,7 @@ test_round_trip_frames()
req.body.value = inst::RequestBodyOpen{open};
inst::Frame own;
own.payload.value = inst::PayloadRequest{req};
const vector<byte> buf = encoded(own);
const vector<byte> buf = wire(own);
dawn::ipc::Decoder dec(buf);
inst::FrameView view{};
CHECK(decode(dec, view));
@@ -224,7 +225,7 @@ test_round_trip_frames()
resp.result.value = inst::ResultDone{};
inst::Frame own;
own.payload.value = inst::PayloadResponse{resp};
const vector<byte> buf = encoded(own);
const vector<byte> buf = wire(own);
dawn::ipc::Decoder dec(buf);
inst::FrameView view{};
CHECK(decode(dec, view));
@@ -237,8 +238,9 @@ test_round_trip_frames()
}
{
inst::Frame own;
own.payload.value = inst::PayloadCancel{inst::Cancel{0x100000002ull}};
const vector<byte> buf = encoded(own);
own.payload.value =
inst::PayloadCancel{dawn::ipc::Cancel{0x100000002ull}};
const vector<byte> buf = wire(own);
dawn::ipc::Decoder dec(buf);
inst::FrameView view{};
CHECK(decode(dec, view));
@@ -264,14 +266,14 @@ test_goldens()
0x01, // session length
0x62, // 'b'
};
check_golden("Hello{1,\"b\"}", encoded(inst::Hello{1, "b"}), kHello);
check_golden("Hello{1,\"b\"}", wire(dawn::ipc::Hello{1, "b"}), kHello);
// HelloReply SessionMismatch: tag i8 = 3, no payload
static constexpr uint8_t kSessionMismatch[] = {0x03};
inst::HelloReply mismatch;
mismatch.value = inst::HelloReplySessionMismatch{};
dawn::ipc::HelloReply mismatch;
mismatch.value = dawn::ipc::HelloReplySessionMismatch{};
check_golden(
"HelloReply SessionMismatch", encoded(mismatch), kSessionMismatch);
"HelloReply SessionMismatch", wire(mismatch), kSessionMismatch);
// Error{NotFound, "x"}: i8 code = 3, u32be len+"x"
static constexpr uint8_t kError[] = {
@@ -283,23 +285,23 @@ test_goldens()
0x78, // 'x'
};
check_golden("Error{NotFound,\"x\"}",
encoded(inst::Error{inst::ErrorCode::NotFound, "x"}), kError);
wire(dawn::ipc::Error{dawn::ipc::ErrorCode::NotFound, "x"}), kError);
}
static void
test_truncation()
{
const vector<byte> full = encoded(inst::Hello{1, "b"});
const vector<byte> full = wire(dawn::ipc::Hello{1, "b"});
CHECK(!full.empty());
for (size_t n = 0; n < full.size(); n++) {
dawn::ipc::Decoder dec(span<const byte>(full.data(), n));
inst::HelloView view{};
dawn::ipc::HelloView view{};
const bool ok = decode(dec, view);
CHECK(!ok);
CHECK(dec.error() == dawn::ipc::DecodeError::Truncated);
}
dawn::ipc::Decoder dec(full);
inst::HelloView view{};
dawn::ipc::HelloView view{};
CHECK(decode(dec, view));
CHECK(dec.remaining() == 0);
CHECK(dec.error() == dawn::ipc::DecodeError::Ok);
@@ -310,10 +312,10 @@ test_truncation()
static void
test_trailing_bytes()
{
vector<byte> buf = encoded(inst::Hello{1, "b"});
vector<byte> buf = wire(dawn::ipc::Hello{1, "b"});
buf.push_back(byte{0x00});
dawn::ipc::Decoder dec(buf);
inst::HelloView view{};
dawn::ipc::HelloView view{};
const bool ok = decode(dec, view);
// Generated Hello decode succeeds and leaves the extra byte.
// Connection/Server require exact consumption.
@@ -347,7 +349,7 @@ test_zero_and_unknown_enum()
for (byte tag : {byte{0}, byte{99}}) {
const byte raw[] = {tag};
dawn::ipc::Decoder dec(raw);
inst::HelloResult value{};
dawn::ipc::HelloResult value{};
CHECK(!decode(dec, value));
}
}
@@ -357,7 +359,7 @@ test_unknown_union_tag()
{
const byte raw[] = {byte{99}};
dawn::ipc::Decoder dec(raw);
inst::HelloReplyView view{};
dawn::ipc::HelloReplyView view{};
CHECK(!decode(dec, view));
}
@@ -394,89 +396,92 @@ test_huge_array_count()
static void
test_daemon_schemas()
{
imaged::DaemonHelloReply hello;
hello.value = imaged::DaemonHelloReplyAccepted{
imaged::DaemonLimits{4096, imaged::kImagedMaxBlobSize}};
auto bytes = encoded(hello);
dawn::ipc::DaemonHelloReply hello;
hello.value = dawn::ipc::DaemonHelloReplyAccepted{
dawn::ipc::DaemonLimits{4096, imaged::kImagedMaxBlobSize}};
auto bytes = wire(hello);
dawn::ipc::Decoder hello_decoder(bytes);
imaged::DaemonHelloReplyView hello_view;
dawn::ipc::DaemonHelloReplyView hello_view;
CHECK(decode(hello_decoder, hello_view));
const auto *accepted =
get_if<imaged::DaemonHelloReplyAcceptedView>(&hello_view.value);
get_if<dawn::ipc::DaemonHelloReplyAcceptedView>(&hello_view.value);
CHECK(accepted != nullptr);
if (accepted)
CHECK(accepted->limits.max_blob_size == imaged::kImagedMaxBlobSize);
imaged::Pixmap pixmap;
dawn::ipc::Pixmap pixmap;
pixmap.width = 2;
pixmap.height = 3;
pixmap.stride = 16;
pixmap.orientation = 6;
pixmap.pixels.value = imaged::BlobShared{48};
pixmap.pixels.value = dawn::ipc::BlobShared{48};
imaged::DecodeResponse response{pixmap, "wuffs", {}, true};
bytes = encoded(response);
bytes = wire(response);
dawn::ipc::Decoder decoder(bytes);
imaged::DecodeResponseView view;
CHECK(decode(decoder, view));
CHECK(holds_alternative<imaged::BlobSharedView>(view.pixmap.pixels.value));
CHECK(get<imaged::BlobSharedView>(view.pixmap.pixels.value).size == 48);
CHECK(
holds_alternative<dawn::ipc::BlobSharedView>(view.pixmap.pixels.value));
CHECK(get<dawn::ipc::BlobSharedView>(view.pixmap.pixels.value).size == 48);
thumbd::ScaleResponse scaled;
scaled.width = 1;
scaled.height = 1;
scaled.rgba8.value =
thumbd::BlobInline{{byte{1}, byte{2}, byte{3}, byte{4}}};
bytes = encoded(scaled);
dawn::ipc::BlobInline{{byte{1}, byte{2}, byte{3}, byte{4}}};
bytes = wire(scaled);
dawn::ipc::Decoder decoder2(bytes);
thumbd::ScaleResponseView scaled_view;
CHECK(decode(decoder2, scaled_view));
CHECK(holds_alternative<thumbd::BlobInlineView>(scaled_view.rgba8.value));
CHECK(
get<thumbd::BlobInlineView>(scaled_view.rgba8.value).bytes.size() == 4);
holds_alternative<dawn::ipc::BlobInlineView>(scaled_view.rgba8.value));
CHECK(
get<dawn::ipc::BlobInlineView>(scaled_view.rgba8.value).bytes.size() ==
4);
}
static void
test_blob_envelope_sizes()
{
imaged::DecodeRequest decode;
decode.data.value = imaged::BlobInline{};
decode.data.value = dawn::ipc::BlobInline{};
imaged::Frame request;
request.payload.value = imaged::PayloadRequest{imaged::Request{1, decode}};
const size_t imaged_overhead = encoded(request).size();
const size_t imaged_overhead = wire(request).size();
get<imaged::PayloadRequest>(request.payload.value)
.request.decode.data.value =
imaged::BlobInline{vector<byte>(123, byte{1})};
CHECK(encoded(request).size() == imaged_overhead + 123);
dawn::ipc::BlobInline{vector<byte>(123, byte{1})};
CHECK(wire(request).size() == imaged_overhead + 123);
imaged::DecodeResponse decoded;
decoded.pixmap.width = 2;
decoded.pixmap.height = 2;
decoded.pixmap.stride = 16;
decoded.pixmap.pixels.value = imaged::BlobInline{};
decoded.pixmap.pixels.value = dawn::ipc::BlobInline{};
decoded.loader = "variable-loader";
decoded.icc = vector<byte>(31, byte{2});
imaged::Frame decoded_frame;
decoded_frame.payload.value = imaged::PayloadResponse{
imaged::Response{2, imaged::Result{imaged::ResultDecoded{decoded}}}};
const size_t decoded_overhead = encoded(decoded_frame).size();
const size_t decoded_overhead = wire(decoded_frame).size();
get<imaged::ResultDecoded>(
get<imaged::PayloadResponse>(decoded_frame.payload.value)
.response.result.value)
.decoded.pixmap.pixels.value =
imaged::BlobInline{vector<byte>(89, byte{3})};
CHECK(encoded(decoded_frame).size() == decoded_overhead + 89);
dawn::ipc::BlobInline{vector<byte>(89, byte{3})};
CHECK(wire(decoded_frame).size() == decoded_overhead + 89);
thumbd::ScaleResponse scale;
scale.width = 1;
scale.height = 1;
scale.rgba8.value = thumbd::BlobInline{};
scale.rgba8.value = dawn::ipc::BlobInline{};
thumbd::Frame response;
response.payload.value = thumbd::PayloadResponse{
thumbd::Response{1, thumbd::Result{thumbd::ResultScaled{scale}}}};
const size_t thumbd_overhead = encoded(response).size();
const size_t thumbd_overhead = wire(response).size();
get<thumbd::PayloadResponse>(response.payload.value).response.result.value =
thumbd::ResultScaled{thumbd::ScaleResponse{
1, 1, thumbd::Blob{thumbd::BlobInline{vector<byte>(77)}}}};
CHECK(encoded(response).size() == thumbd_overhead + 77);
1, 1, dawn::ipc::Blob{dawn::ipc::BlobInline{vector<byte>(77)}}}};
CHECK(wire(response).size() == thumbd_overhead + 77);
}
int
tools/dnimage.cpp
+26 -56
View File
@@ -25,14 +25,6 @@ namespace proto = dawn::ipc::imaged;
// --- IPC ---------------------------------------------------------------------
static vector<byte>
encoded(const proto::Frame &frame)
{
vector<byte> out;
dawn::ipc::Encoder encoder(out);
encode(frame, encoder);
return out;
}
static bool
receive(dawn::ipc::Channel &channel, proto::FrameView &view,
vector<byte> &storage, vector<dawn::ipc::Handle> &handles,
@@ -76,9 +68,9 @@ main()
}
proto::Frame hello;
hello.payload.value =
proto::PayloadHello{proto::Hello{proto::kImagedProtocolVersion, ""}};
if (!channel.send(encoded(hello), budget, {})) {
hello.payload.value = proto::PayloadHello{
dawn::ipc::Hello{proto::kImagedProtocolVersion, ""}};
if (!channel.send(dawn::ipc::encoded(hello), budget, {})) {
fprintf(stderr, "dnimage: handshake failed\n");
return 1;
}
@@ -93,7 +85,7 @@ main()
return 1;
}
const auto *accepted = get_if<proto::DaemonHelloReplyAcceptedView>(
const auto *accepted = get_if<dawn::ipc::DaemonHelloReplyAcceptedView>(
&get<proto::PayloadHelloReplyView>(reply.payload.value)
.hello_reply.value);
if (!accepted || !accepted->limits.max_payload_size ||
@@ -113,35 +105,25 @@ main()
}
proto::DecodeRequest decode;
dawn::ipc::SharedMemory request_memory;
dawn::ipc::Handle request_handle;
span<const dawn::ipc::Handle> request_handles;
decode.first_frame_only = true;
decode.data.value = proto::BlobInline{};
proto::Frame request;
request.payload.value = proto::PayloadRequest{proto::Request{1, decode}};
const size_t inline_overhead = encoded(request).size();
const auto encode_request = [&] {
proto::Frame request;
request.payload.value =
proto::PayloadRequest{proto::Request{1, decode}};
return dawn::ipc::encoded(request);
};
vector<byte> request_wire;
if (inline_overhead <= max_payload &&
input.size() <= max_payload - inline_overhead) {
decode.data.value = proto::BlobInline{std::move(input)};
request.payload.value =
proto::PayloadRequest{proto::Request{1, std::move(decode)}};
request_wire = encoded(request);
} else {
request_memory =
dawn::ipc::SharedMemory::copy(input.data(), input.size());
if (!request_memory.ok())
return 1;
decode.data.value = proto::BlobShared{input.size()};
request_handle = request_memory.handle();
request_handles = span(&request_handle, 1);
request.payload.value =
proto::PayloadRequest{proto::Request{1, std::move(decode)}};
request_wire = encoded(request);
dawn::ipc::SharedMemory request_memory;
if (dawn::ipc::place_blob(decode.data, input, max_payload, encode_request,
request_wire, request_memory)) {
fprintf(stderr, "dnimage: request failed\n");
return 1;
}
if (request_wire.size() > max_payload ||
!channel.send(request_wire, budget, request_handles)) {
const dawn::ipc::Handle request_handle = request_memory.handle();
if (!channel.send(request_wire, budget,
request_memory.ok() ? span(&request_handle, 1)
: span<const dawn::ipc::Handle>())) {
fprintf(stderr, "dnimage: request failed\n");
return 1;
}
@@ -184,25 +166,13 @@ main()
return 1;
}
const uint8_t *pixels = nullptr;
span<const byte> bytes;
dawn::ipc::SharedMemory response_memory;
if (auto in = get_if<proto::BlobInlineView>(&pixmap.pixels.value)) {
if (!owned_handles.empty() || in->bytes.size() > max_blob ||
uint64_t(pixmap.stride) * pixmap.height > in->bytes.size())
return 1;
pixels = reinterpret_cast<const uint8_t *>(in->bytes.data());
} else {
auto shared = get<proto::BlobSharedView>(pixmap.pixels.value);
if (owned_handles.size() != 1 || !shared.size ||
shared.size > max_blob ||
uint64_t(pixmap.stride) * pixmap.height > shared.size)
return 1;
response_memory =
dawn::ipc::SharedMemory::map(owned_handles.take(0), shared.size);
if (!response_memory.ok())
return 1;
pixels = response_memory.data();
}
if (dawn::ipc::take_blob(
pixmap.pixels, owned_handles, max_blob, bytes, response_memory) ||
uint64_t(pixmap.stride) * pixmap.height > bytes.size())
return 1;
const auto *pixels = reinterpret_cast<const uint8_t *>(bytes.data());
auto cmm = make_shared<dawn::Cmm>();
auto icc = cmm->get_profile_sRGB(false)->to_bytes();
tools/dnthumb.cpp
+28 -64
View File
@@ -27,14 +27,6 @@ namespace proto = dawn::ipc::thumbd;
// --- Helpers -----------------------------------------------------------------
static vector<byte>
encoded(const proto::Frame &frame)
{
vector<byte> out;
dawn::ipc::Encoder encoder(out);
encode(frame, encoder);
return out;
}
static bool
parse_size(string_view value, uint32_t &w, uint32_t &h)
{
@@ -137,9 +129,9 @@ main(int argc, char **argv)
}
proto::Frame hello;
hello.payload.value =
proto::PayloadHello{proto::Hello{proto::kThumbdProtocolVersion, ""}};
if (!channel.send(encoded(hello), budget, {}))
hello.payload.value = proto::PayloadHello{
dawn::ipc::Hello{proto::kThumbdProtocolVersion, ""}};
if (!channel.send(dawn::ipc::encoded(hello), budget, {}))
return 1;
vector<byte> wire;
vector<dawn::ipc::Handle> reply_handles;
@@ -152,7 +144,7 @@ main(int argc, char **argv)
return 1;
}
const auto *accepted = get_if<proto::DaemonHelloReplyAcceptedView>(
const auto *accepted = get_if<dawn::ipc::DaemonHelloReplyAcceptedView>(
&get<proto::PayloadHelloReplyView>(reply.payload.value)
.hello_reply.value);
if (!accepted || !accepted->limits.max_payload_size ||
@@ -175,45 +167,28 @@ main(int argc, char **argv)
scale_request.source.orientation = orientation;
scale_request.out_width = out_w;
scale_request.out_height = out_h;
dawn::ipc::SharedMemory request_memory;
dawn::ipc::Handle request_handle;
span<const dawn::ipc::Handle> request_handles;
if (image->data.empty() || image->data.size() > max_blob) {
fprintf(stderr, "dnthumb: image is too large\n");
return 1;
}
scale_request.source.pixels.value = proto::BlobInline{};
proto::Frame request;
request.payload.value =
proto::PayloadRequest{proto::Request{1, scale_request}};
const size_t inline_overhead = encoded(request).size();
const auto encode_request = [&] {
proto::Frame request;
request.payload.value =
proto::PayloadRequest{proto::Request{1, scale_request}};
return dawn::ipc::encoded(request);
};
vector<byte> request_wire;
if (inline_overhead <= max_payload &&
image->data.size() <= max_payload - inline_overhead) {
proto::BlobInline inline_blob;
inline_blob.bytes.assign(
reinterpret_cast<const byte *>(image->data.data()),
reinterpret_cast<const byte *>(
image->data.data() + image->data.size()));
scale_request.source.pixels.value = std::move(inline_blob);
request.payload.value =
proto::PayloadRequest{proto::Request{1, std::move(scale_request)}};
request_wire = encoded(request);
} else {
request_memory = dawn::ipc::SharedMemory::copy(
image->data.data(), image->data.size());
if (!request_memory.ok())
return 1;
scale_request.source.pixels.value =
proto::BlobShared{image->data.size()};
request_handle = request_memory.handle();
request_handles = span(&request_handle, 1);
request.payload.value =
proto::PayloadRequest{proto::Request{1, std::move(scale_request)}};
request_wire = encoded(request);
}
if (request_wire.size() > max_payload ||
!channel.send(request_wire, budget, request_handles))
dawn::ipc::SharedMemory request_memory;
if (dawn::ipc::place_blob(scale_request.source.pixels,
as_bytes(span(image->data)), max_payload, encode_request,
request_wire, request_memory))
return 1;
const dawn::ipc::Handle request_handle = request_memory.handle();
if (!channel.send(request_wire, budget,
request_memory.ok() ? span(&request_handle, 1)
: span<const dawn::ipc::Handle>()))
return 1;
vector<dawn::ipc::Handle> handles;
@@ -253,25 +228,14 @@ main(int argc, char **argv)
return 1;
}
span<const uint8_t> rgba;
span<const byte> bytes;
dawn::ipc::SharedMemory response_memory;
if (auto in = get_if<proto::BlobInlineView>(&scaled->scaled.rgba8.value)) {
if (!owned_handles.empty() || in->bytes.size() != rgba_size ||
in->bytes.size() > max_blob)
return 1;
rgba = {reinterpret_cast<const uint8_t *>(in->bytes.data()),
in->bytes.size()};
} else {
auto shared = get<proto::BlobSharedView>(scaled->scaled.rgba8.value);
if (owned_handles.size() != 1 || shared.size != rgba_size ||
shared.size > max_blob)
return 1;
response_memory =
dawn::ipc::SharedMemory::map(owned_handles.take(0), shared.size);
if (!response_memory.ok())
return 1;
rgba = {response_memory.data(), response_memory.size()};
}
if (dawn::ipc::take_blob(scaled->scaled.rgba8, owned_handles, max_blob,
bytes, response_memory) ||
bytes.size() != rgba_size)
return 1;
const span<const uint8_t> rgba(
reinterpret_cast<const uint8_t *>(bytes.data()), bytes.size());
vector<uint8_t> icc = image->effective_profile
? image->effective_profile->to_bytes()