WIP: Finish the IPC story

This commit is contained in:
p
2026-09-08 09:39:24 +02:00
parent 5ce8d21943
commit 0df1955435
32 changed files with 3239 additions and 288 deletions
+5
View File
@@ -92,6 +92,7 @@ pkg_check_modules(WEBP REQUIRED IMPORTED_TARGET libwebp)
pkg_check_modules(WEBPDEMUX REQUIRED IMPORTED_TARGET libwebpdemux)
pkg_check_modules(WEBPMUX REQUIRED IMPORTED_TARGET libwebpmux)
pkg_check_modules(ZLIB REQUIRED IMPORTED_TARGET zlib)
pkg_check_modules(LIBPNG IMPORTED_TARGET libpng)
# Qt6 SVG cannot render our icons correctly, so we hard depend on resvg.
pkg_check_modules(RESVG REQUIRED IMPORTED_TARGET resvg)
@@ -244,7 +245,11 @@ if (CMAKE_CXX_COMPILER_ID MATCHES "GNU|Clang")
endif()
add_subdirectory(libdn)
add_subdirectory(dnimaged)
add_subdirectory(dnthumbd)
if (LIBPNG_FOUND)
add_subdirectory(tools)
endif()
add_subdirectory(dn)
if (BUILD_TESTING)
+5 -4
View File
@@ -72,7 +72,7 @@ map_open_error(OpenResult r)
struct InstanceHost::Impl {
Impl(dawn::ipc::Listener listener, App &app, const QString &session,
InstanceHost *host);
void watch_read(uint64_t id, dawn::ipc::Waitable w);
bool watch_read(uint64_t id, dawn::ipc::Waitable w);
void watch_write(uint64_t id, dawn::ipc::Waitable w, bool enable);
void unwatch(uint64_t id);
void on_request(dawn::ipc::instance::Call call,
@@ -96,7 +96,7 @@ InstanceHost::Impl::Impl(dawn::ipc::Listener listener, App &app,
on_request(std::move(call), req);
};
cfg.watch_read = [this](uint64_t id, dawn::ipc::Waitable w) {
watch_read(id, w);
return watch_read(id, w);
};
cfg.unwatch = [this](uint64_t id) { unwatch(id); };
cfg.watch_write = [this](uint64_t id, dawn::ipc::Waitable w, bool enable) {
@@ -110,15 +110,16 @@ InstanceHost::Impl::Impl(dawn::ipc::Listener listener, App &app,
[this] { this->server_->poll_listen(); });
}
void
bool
InstanceHost::Impl::watch_read(uint64_t id, dawn::ipc::Waitable w)
{
if (this->reads_.contains(id))
return;
return true;
auto *n = make_watch(w, false, this->host_);
QObject::connect(n, &Watch::activated, this->host_,
[this, id] { this->server_->poll_read(id); });
this->reads_[id] = n;
return true;
}
void
+11
View File
@@ -0,0 +1,11 @@
add_executable(dnimaged
main.cpp
)
target_link_libraries(dnimaged PRIVATE
dawn::libdn
)
target_include_directories(dnimaged PRIVATE
${PROJECT_SOURCE_DIR}/libdn
)
+396
View File
@@ -0,0 +1,396 @@
//
// main.cpp: out-of-process image decoder probe
//
// Copyright The Dawn Authors
// 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 <thread>
#include <unordered_set>
#include <variant>
#include <vector>
using namespace std;
namespace proto = dawn::ipc::imaged;
// --- Jobs --------------------------------------------------------------------
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)
{
vector<byte> out;
dawn::ipc::Encoder encoder(out);
encode(frame, encoder);
return out;
}
namespace
{
struct Job {
uint64_t connection = 0, request = 0;
size_t retained_bytes = 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;
};
} // namespace
static bool
admit(State &state, Job &job, size_t bytes)
{
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 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::OpenContext context;
context.cmm = cmm;
context.first_frame_only = job.first_frame_only;
context.screen_profile = job.target_icc.empty()
? cmm->get_profile_sRGB(false)
: cmm->get_profile(span<const uint8_t>(
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;
}
dawn::Error error;
const uint8_t *data = job.shared.ok()
? job.shared.data()
: reinterpret_cast<const uint8_t *>(job.data.data());
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,
error.message.empty() ? "decode failed" : error.message,
proto::ErrorCode::InvalidArgument);
return done;
}
if (image->data.empty() || image->data.size() > kMaxBlob) {
done.frame = error_frame(job.request, "decoded pixmap is too large",
proto::ErrorCode::InvalidArgument);
return done;
}
proto::DecodeResponse response;
response.pixmap.width = image->width;
response.pixmap.height = image->height;
response.pixmap.stride = image->stride;
response.pixmap.orientation = int32_t(image->orientation);
response.loader = image->loader ? image->loader : "";
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();
}
}
// --- Server ------------------------------------------------------------------
static bool
handle_payload(State &state, uint64_t connection, span<const byte> payload,
vector<dawn::ipc::Handle> attachments)
{
dawn::ipc::OwnedHandles owned(std::move(attachments));
dawn::ipc::Decoder decoder(payload);
proto::FrameView frame;
if (!decode(decoder, frame) || decoder.remaining())
return false;
if (auto hello = get_if<proto::PayloadHelloView>(&frame.payload.value)) {
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}};
state.greeted.insert(connection);
}
proto::Frame out;
out.payload.value = proto::PayloadHelloReply{reply};
return state.server->send(connection, encode_frame(out), {});
}
if (!state.greeted.contains(connection))
return false;
auto request = get_if<proto::PayloadRequestView>(&frame.payload.value);
if (!request)
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)),
{});
};
const auto busy = [&] {
return state.server->send(connection,
encode_frame(error_frame(request->request.id,
"image service is busy", proto::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))
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()
{
auto endpoint = dawn::ipc::Endpoint::listen(kService);
if (endpoint.status != dawn::ipc::Endpoint::ListenStatus::Ok) {
fprintf(stderr, "dnimaged: %s\n",
endpoint.status == dawn::ipc::Endpoint::ListenStatus::InUse
? "already running"
: "cannot listen");
return 1;
}
State state;
dawn::ipc::ServerCore::Config config;
config.on_payload = [&](uint64_t connection, span<const byte> payload,
vector<dawn::ipc::Handle> attachments) {
return handle_payload(
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();
}
+333 -177
View File
@@ -1,228 +1,384 @@
//
// main.cpp: dnthumbd thumbnail daemon entry point
// main.cpp: out-of-process Vulkan scaler probe
//
// Copyright The Dawn Authors
// 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 <atomic>
#include <condition_variable>
#include <cstdio>
#include <cstdlib>
#include <filesystem>
#include <getopt.h>
#include <deque>
#include <mutex>
#include <queue>
#include <string>
#include <thread>
#include <vector>
#include <unordered_set>
#include <variant>
using namespace std;
namespace fs = filesystem;
namespace proto = dawn::ipc::thumbd;
static mutex io_mu;
static atomic<bool> any_fail{false};
// --- Jobs --------------------------------------------------------------------
static void
log_err(const string &msg)
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)
{
lock_guard lock(io_mu);
fprintf(stderr, "%s\n", msg.c_str());
vector<byte> out;
dawn::ipc::Encoder encoder(out);
encode(frame, encoder);
return out;
}
static void
fit_size(uint32_t w, uint32_t h, uint32_t *out_w, uint32_t *out_h)
static proto::Frame
error_frame(uint64_t id, string message, proto::ErrorCode code)
{
const float scale = min(1.0f, min(512.0f / float(w), 256.0f / float(h)));
*out_w = max(1u, uint32_t(float(w) * scale + 0.5f));
*out_h = max(1u, uint32_t(float(h) * scale + 0.5f));
}
static fs::path
thumb_path(const fs::path &input)
{
const fs::path dir = input.parent_path();
const string stem = input.stem().string();
return dir / (stem + ".thumb.webp");
}
static bool
process_one(const string &path, dawn::ScaleScaler *scaler)
{
dawn::OpenContext ctx;
ctx.uri = path;
// One Cmm per job — do not share get_default() across worker threads.
ctx.cmm = make_shared<dawn::Cmm>();
ctx.screen_profile = ctx.cmm->get_profile_sRGB(false);
ctx.first_frame_only = true;
dawn::Error error;
dawn::ImagePtr image = dawn::open(ctx, &error);
if (!image) {
log_err(path + ": " +
(error.message.empty() ? "open failed" : error.message));
return false;
}
if (image->width == 0 || image->height == 0) {
log_err(path + ": empty image");
return false;
}
// The scaler orients as it filters; fit against the oriented size.
const dawn::Orientation orientation =
dawn::orientation_or_0(image->orientation);
uint32_t disp_w = 0, disp_h = 0;
dawn::orientation_display_size(
image->width, image->height, orientation, &disp_w, &disp_h);
uint32_t out_w = 0, out_h = 0;
fit_size(disp_w, disp_h, &out_w, &out_h);
dawn::ScaleOutput scaled;
string vk_err;
if (!scaler->scale(image->width, image->height, image->data.data(),
image->stride, out_w, out_h, orientation, &scaled, &vk_err)) {
log_err(path + ": scale failed: " + vk_err);
return false;
}
vector<uint8_t> webp;
string enc_err;
if (!dawn::encode_thumbnail_webp(scaled.width, scaled.height,
scaled.rgba8.data(), size_t(scaled.width) * 4, &webp, &enc_err)) {
log_err(path + ": encode failed: " + enc_err);
return false;
}
const fs::path out_path = thumb_path(path);
FILE *f = fopen(out_path.string().c_str(), "wb");
if (!f) {
log_err(out_path.string() + ": cannot write");
return false;
}
const size_t n = fwrite(webp.data(), 1, webp.size(), f);
fclose(f);
if (n != webp.size()) {
log_err(out_path.string() + ": short write");
return false;
}
lock_guard lock(io_mu);
printf("%s -> %s (%ux%u)\n", path.c_str(), out_path.string().c_str(), out_w,
out_h);
return true;
proto::Frame frame;
frame.payload.value = proto::PayloadResponse{proto::Response{id,
proto::Result{
proto::ResultError{proto::Error{code, std::move(message)}}}}};
return frame;
}
namespace
{
struct JobQueue {
struct Job {
uint64_t connection = 0, request = 0;
size_t retained_bytes = 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;
queue<string> paths;
bool done = false;
void push(string path)
{
lock_guard lock(mu);
paths.push(std::move(path));
cv.notify_one();
}
void finish()
{
lock_guard lock(mu);
done = true;
cv.notify_all();
}
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;
};
} // namespace
static void
worker(JobQueue *jobs, dawn::ScaleScaler *scaler)
static bool
admit(State &state, Job &job, size_t bytes)
{
while (true) {
string path;
{
unique_lock lock(jobs->mu);
jobs->cv.wait(
lock, [&] { return !jobs->paths.empty() || jobs->done; });
if (jobs->paths.empty() && jobs->done)
return;
path = std::move(jobs->paths.front());
jobs->paths.pop();
}
if (!process_one(path, scaler))
any_fail = true;
}
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
usage(const char *argv0)
release(State &state, uint64_t connection, size_t bytes)
{
fprintf(stderr, "Usage: %s [-j N] IMAGE...\n", argv0);
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::ScaleOutput output;
string error;
const uint8_t *pixels = job.shared.ok()
? job.shared.data()
: reinterpret_cast<const uint8_t *>(job.inline_pixels.data());
if (!scaler->scale(job.width, job.height, pixels, job.stride, job.out_width,
job.out_height,
dawn::orientation_or_0(dawn::Orientation(job.orientation)), &output,
&error)) {
done.frame =
error_frame(job.request, error.empty() ? "scale failed" : error,
proto::ErrorCode::InvalidArgument);
return done;
}
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();
}
}
// --- Server ------------------------------------------------------------------
static bool
handle_payload(State &state, uint64_t connection, span<const byte> payload,
vector<dawn::ipc::Handle> attachments)
{
dawn::ipc::OwnedHandles owned(std::move(attachments));
dawn::ipc::Decoder decoder(payload);
proto::FrameView frame;
if (!decode(decoder, frame) || decoder.remaining())
return false;
if (auto hello = get_if<proto::PayloadHelloView>(&frame.payload.value)) {
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}};
state.greeted.insert(connection);
}
proto::Frame out;
out.payload.value = proto::PayloadHelloReply{reply};
return state.server->send(connection, encode_frame(out), {});
}
if (!state.greeted.contains(connection))
return false;
auto request = get_if<proto::PayloadRequestView>(&frame.payload.value);
if (!request)
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)),
{});
};
const auto busy = [&] {
return state.server->send(connection,
encode_frame(error_frame(request->request.id,
"thumbnail service is busy", proto::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)
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)
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))
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(int argc, char **argv)
main()
{
unsigned nthreads = thread::hardware_concurrency();
if (nthreads < 1)
nthreads = 1;
const option kLong[] = {
{"help", no_argument, nullptr, 'h'},
{nullptr, 0, nullptr, 0},
};
int c;
while ((c = getopt_long(argc, argv, "j:h", kLong, nullptr)) != -1) {
switch (c) {
case 'j': {
const long n = strtol(optarg, nullptr, 10);
if (n > 0)
nthreads = unsigned(n);
break;
}
case 'h':
usage(argv[0]);
return 0;
default:
usage(argv[0]);
return 1;
}
}
vector<string> paths;
for (int i = optind; i < argc; i++)
paths.push_back(argv[i]);
if (paths.empty()) {
usage(argv[0]);
auto endpoint = dawn::ipc::Endpoint::listen("thumbd");
if (endpoint.status != dawn::ipc::Endpoint::ListenStatus::Ok) {
fprintf(stderr, "dnthumbd: %s\n",
endpoint.status == dawn::ipc::Endpoint::ListenStatus::InUse
? "already running"
: "cannot listen");
return 1;
}
dawn::ScaleScaler scaler;
string err;
if (!scaler.init(&err)) {
fprintf(stderr, "Vulkan init failed: %s\n", err.c_str());
string init_error;
if (!scaler.init(&init_error)) {
fprintf(stderr, "dnthumbd: %s\n", init_error.c_str());
return 1;
}
JobQueue jobs;
for (const string &p : paths)
jobs.push(p);
jobs.finish();
State state;
dawn::ipc::ServerCore::Config config;
config.on_payload = [&](uint64_t connection, span<const byte> payload,
vector<dawn::ipc::Handle> attachments) {
return handle_payload(
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); };
vector<thread> pool;
pool.reserve(nthreads);
for (unsigned i = 0; i < nthreads; i++)
pool.emplace_back(worker, &jobs, &scaler);
for (thread &t : pool)
t.join();
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;
return any_fail ? 1 : 0;
thread thread(worker, &state, &scaler);
state.loop.run();
{
lock_guard lock(state.mu);
state.stop = true;
}
state.cv.notify_all();
thread.join();
}
+37
View File
@@ -44,3 +44,40 @@ struct Error {
struct Cancel {
u64 id;
};
// Where a large payload lives. Inline stays within the negotiated maximum
// payload; shared means the Nth attachment on the frame carrying it.
union Blob switch (enum BlobKind {
INLINE,
SHARED,
} kind) {
case INLINE:
u8 bytes<>;
case SHARED:
u64 size;
};
union DaemonHelloReply switch (enum DaemonHelloResult {
ACCEPTED,
VERSION_MISMATCH,
SESSION_MISMATCH,
} result) {
case ACCEPTED:
struct DaemonLimits {
u32 max_payload_size;
u64 max_blob_size;
} limits;
case VERSION_MISMATCH:
u32 server_protocol_version;
case SESSION_MISMATCH:
void;
};
// Working format: BGRA_PREMUL_4X16LE, stride in bytes.
struct Pixmap {
u32 width;
u32 height;
u32 stride;
i32 orientation; // Exif, 0 = unknown; carried, not applied
Blob pixels;
};
+58
View File
@@ -0,0 +1,58 @@
//
// imaged.lxdr: image decoder daemon protocol
//
// Copyright The Dawn Authors
// SPDX-License-Identifier: MPL-2.0
//
const IMAGED_PROTOCOL_VERSION = 1;
const IMAGED_MAX_BLOB_SIZE = 1073741824;
struct DecodeRequest {
Blob data;
u8 target_icc<>; // empty means sRGB
bool first_frame_only;
};
struct DecodeResponse {
Pixmap pixmap;
string loader;
u8 icc<>; // the file's own profile blob, if any
bool profile_assumed;
};
struct Request {
u64 id;
DecodeRequest decode;
};
struct Response {
u64 id;
union Result switch (enum ResponseKind {
DECODED,
ERROR,
} kind) {
case DECODED:
DecodeResponse decoded;
case ERROR:
Error error;
} result;
};
struct Frame {
union Payload switch (enum FrameKind {
HELLO,
HELLO_REPLY,
REQUEST,
RESPONSE,
} kind) {
case HELLO:
Hello hello;
case HELLO_REPLY:
DaemonHelloReply hello_reply;
case REQUEST:
Request request;
case RESPONSE:
Response response;
} payload;
};
+57
View File
@@ -0,0 +1,57 @@
//
// thumbd.lxdr: thumbnail scaler daemon protocol
//
// Copyright The Dawn Authors
// SPDX-License-Identifier: MPL-2.0
//
const THUMBD_PROTOCOL_VERSION = 1;
const THUMBD_MAX_BLOB_SIZE = 1073741824;
struct ScaleRequest {
Pixmap source;
u32 out_width;
u32 out_height;
};
struct ScaleResponse {
u32 width;
u32 height;
Blob rgba8; // straight, non-premultiplied, stride = width * 4
};
struct Request {
u64 id;
ScaleRequest scale;
};
struct Response {
u64 id;
union Result switch (enum ResponseKind {
SCALED,
ERROR,
} kind) {
case SCALED:
ScaleResponse scaled;
case ERROR:
Error error;
} result;
};
struct Frame {
union Payload switch (enum FrameKind {
HELLO,
HELLO_REPLY,
REQUEST,
RESPONSE,
} kind) {
case HELLO:
Hello hello;
case HELLO_REPLY:
DaemonHelloReply hello_reply;
case REQUEST:
Request request;
case RESPONSE:
Response response;
} payload;
};
+42 -4
View File
@@ -87,12 +87,25 @@ add_library(dawn::libdn ALIAS libdn)
if (WIN32)
target_sources(libdn PRIVATE
config-windows.cpp ipc-windows.cpp ipc-rpc.cpp ipc-instance.cpp)
config-windows.cpp
ipc-instance.cpp
ipc-loop-windows.cpp
ipc-rpc.cpp
ipc-shm-windows.cpp
ipc-windows.cpp
)
target_link_libraries(libdn PRIVATE advapi32)
elseif (APPLE)
enable_language(OBJCXX)
find_library(DAWN_COREFOUNDATION_FRAMEWORK CoreFoundation REQUIRED)
target_sources(libdn PRIVATE config-macos.mm)
target_sources(libdn PRIVATE
config-macos.mm
ipc-instance.cpp
ipc-loop-unix.cpp
ipc-rpc.cpp
ipc-shm-unix.cpp
ipc-unix.cpp
)
target_link_libraries(libdn PRIVATE "${DAWN_COREFOUNDATION_FRAMEWORK}")
if (DAWN_WITH_IMAGEIO OR DAWN_WITH_CGPDF)
@@ -111,7 +124,13 @@ elseif (APPLE)
endif()
elseif (NOT APPLE)
target_sources(libdn PRIVATE
config-unix.cpp ipc-unix.cpp ipc-rpc.cpp ipc-instance.cpp)
config-unix.cpp
ipc-instance.cpp
ipc-loop-unix.cpp
ipc-rpc.cpp
ipc-shm-unix.cpp
ipc-unix.cpp
)
endif()
# Generated LXDR header; ipc-instance.cpp binds it to the RPC core, and
@@ -123,9 +142,28 @@ lxdr_generate(
${PROJECT_SOURCE_DIR}/ipc/common.lxdr
${PROJECT_SOURCE_DIR}/ipc/instance.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
)
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
)
add_custom_target(dn-tiff-tables
DEPENDS ${CMAKE_CURRENT_BINARY_DIR}/tiff-tables.h)
add_dependencies(libdn dn-tiff-tables lxdr-instance_lxdr_hpp)
add_dependencies(libdn
dn-tiff-tables
lxdr-imaged_lxdr_hpp
lxdr-instance_lxdr_hpp
lxdr-thumbd_lxdr_hpp
)
target_include_directories(libdn
PUBLIC
+12 -9
View File
@@ -117,7 +117,7 @@ Call::complete(Result result)
frame.payload.value = PayloadResponse{std::move(response)};
vector<byte> buf;
if (encode_frame(frame, buf))
(void) s->core->send(s->conn, buf);
(void) s->core->send(s->conn, buf, {});
s->owner->retire(s->conn, s->id);
}
@@ -127,7 +127,10 @@ Server::Server(Listener listener, Config cfg) : cfg_(std::move(cfg))
{
ServerCore::Config core;
core.max_payload_size = this->cfg_.max_payload_size;
core.on_payload = [this](uint64_t id, span<const byte> payload) {
core.on_payload = [this](uint64_t id, span<const byte> payload,
vector<Handle> attachments) {
for (Handle h : attachments)
close_handle(h);
return this->on_payload(id, payload);
};
core.on_closed = [this](uint64_t id) { this->on_closed(id); };
@@ -172,7 +175,7 @@ bool
Server::send_frame(uint64_t id, const Frame &frame)
{
vector<byte> buf;
return encode_frame(frame, buf) && this->core_->send(id, buf);
return encode_frame(frame, buf) && this->core_->send(id, buf, {});
}
void
@@ -333,9 +336,9 @@ Client::connect(
Frame frame;
frame.payload.value = PayloadHello{std::move(hello)};
vector<byte> buf;
if (!encode_frame(frame, buf) || !chan.send(buf, left))
if (!encode_frame(frame, buf) || !chan.send(buf, left, {}))
return fail(HelloStatus::Unavailable, "send");
if (!chan.recv(buf, left))
if (!chan.recv(buf, left, nullptr))
return fail(HelloStatus::Unavailable, "no reply");
Decoder dec(buf);
@@ -378,7 +381,7 @@ Client::cancel(uint64_t id)
frame.payload.value = PayloadCancel{Cancel{id}};
vector<byte> buf;
if (encode_frame(frame, buf))
(void) this->chan_.send(buf, left);
(void) this->chan_.send(buf, left, {});
}
bool
@@ -394,11 +397,11 @@ Client::call(const Request &req, Received<ResponseView> &out, Error *error,
Frame frame;
frame.payload.value = PayloadRequest{req};
vector<byte> buf;
if (!encode_frame(frame, buf) || !this->chan_.send(buf, left)) {
if (!encode_frame(frame, buf) || !this->chan_.send(buf, left, {})) {
set_internal(error);
return false;
}
if (!this->chan_.recv(buf, left)) {
if (!this->chan_.recv(buf, left, nullptr)) {
this->cancel(req.id);
set_internal(error);
return false;
@@ -445,7 +448,7 @@ Client::open(const vector<string> &urls, string_view activation_token,
open_req.browse = browse;
Request req;
req.id = ++this->last_id_;
req.id = this->last_id_ += 1;
req.body.value = RequestBodyOpen{std::move(open_req)};
Received<ResponseView> response;
+1 -1
View File
@@ -88,7 +88,7 @@ public:
// and is only valid until this returns; call is not.
std::function<void(Call call, const RequestView &req)> on_request;
// Event-loop hooks, passed straight through to ServerCore.
std::function<void(uint64_t id, Waitable w)> watch_read;
std::function<bool(uint64_t id, Waitable w)> watch_read;
std::function<void(uint64_t id)> unwatch;
std::function<void(uint64_t id, Waitable w, bool enable)> watch_write;
};
+122
View File
@@ -0,0 +1,122 @@
//
// ipc-loop-unix.cpp: poll-based service event loop
//
// Copyright The Dawn Authors
// SPDX-License-Identifier: MPL-2.0
//
#include "ipc-loop.hpp"
#include <cerrno>
#include <fcntl.h>
#include <poll.h>
#include <unistd.h>
#include <unordered_map>
#include <vector>
using namespace std;
namespace dawn::ipc
{
struct Loop::Impl {
struct Watch {
int fd = -1;
bool read = false;
bool write = false;
};
unordered_map<uint64_t, Watch> watches;
int wake_pipe[2] = {-1, -1};
};
Loop::Loop() : impl_(make_unique<Impl>())
{
#if defined __linux__
if (pipe2(impl_->wake_pipe, O_CLOEXEC | O_NONBLOCK) != 0)
impl_->wake_pipe[0] = impl_->wake_pipe[1] = -1;
#else
if (pipe(impl_->wake_pipe) == 0)
for (int fd : impl_->wake_pipe) {
fcntl(fd, F_SETFD, FD_CLOEXEC);
fcntl(fd, F_SETFL, fcntl(fd, F_GETFL) | O_NONBLOCK);
}
#endif
}
Loop::~Loop()
{
for (int fd : impl_->wake_pipe)
if (fd >= 0)
::close(fd);
}
bool
Loop::watch_read(uint64_t id, Waitable w)
{
auto &watch = impl_->watches[id];
watch.fd = int(w);
watch.read = true;
return true;
}
void
Loop::watch_write(uint64_t id, Waitable w, bool enable)
{
auto &watch = impl_->watches[id];
watch.fd = int(w);
watch.write = enable;
}
void
Loop::unwatch(uint64_t id)
{
impl_->watches.erase(id);
}
void
Loop::wake()
{
if (impl_->wake_pipe[1] >= 0) {
const char byte = 0;
(void) ::write(impl_->wake_pipe[1], &byte, 1);
}
}
void
Loop::run()
{
while (true) {
vector<pollfd> fds{{impl_->wake_pipe[0], POLLIN, 0}};
vector<uint64_t> ids{0};
for (const auto &[id, watch] : impl_->watches) {
short events =
(watch.read ? POLLIN : 0) | (watch.write ? POLLOUT : 0);
fds.push_back({watch.fd, events, 0});
ids.push_back(id);
}
const int n = poll(fds.data(), fds.size(), -1);
if (n < 0) {
if (errno == EINTR)
continue;
break;
}
if (fds[0].revents) {
char buf[64];
while (::read(fds[0].fd, buf, sizeof buf) > 0) {
}
if (on_wake)
on_wake();
}
for (size_t i = 1; i < fds.size(); i++) {
const short re = fds[i].revents;
if ((re & (POLLIN | POLLHUP | POLLERR)) && on_read)
on_read(ids[i]);
if ((re & (POLLOUT | POLLERR)) && on_write &&
impl_->watches.contains(ids[i]))
on_write(ids[i]);
}
}
}
} // namespace dawn::ipc
+114
View File
@@ -0,0 +1,114 @@
//
// ipc-loop-windows.cpp: WaitForMultipleObjects service event loop
//
// Copyright The Dawn Authors
// SPDX-License-Identifier: MPL-2.0
//
#include "ipc-loop.hpp"
#define WIN32_LEAN_AND_MEAN
#include <windows.h>
#include <unordered_map>
#include <vector>
using namespace std;
namespace dawn::ipc
{
struct Loop::Impl {
struct Watch {
HANDLE read = nullptr;
HANDLE write = nullptr;
bool write_on = false;
};
unordered_map<uint64_t, Watch> watches;
HANDLE wake = CreateEventW(nullptr, TRUE, FALSE, nullptr);
};
Loop::Loop() : impl_(make_unique<Impl>())
{
}
Loop::~Loop()
{
if (impl_->wake)
CloseHandle(impl_->wake);
}
bool
Loop::watch_read(uint64_t id, Waitable w)
{
if (!impl_->watches.contains(id)) {
size_t reserved = 1;
for (const auto &[watch_id, watch] : impl_->watches)
if (watch.read || watch.write)
reserved += watch_id == kListener ? 1 : 2;
reserved += id == kListener ? 1 : 2;
if (reserved > MAXIMUM_WAIT_OBJECTS)
return false;
}
impl_->watches[id].read = (HANDLE) w;
return true;
}
void
Loop::watch_write(uint64_t id, Waitable w, bool enable)
{
auto &watch = impl_->watches[id];
watch.write = (HANDLE) w;
watch.write_on = enable;
}
void
Loop::unwatch(uint64_t id)
{
impl_->watches.erase(id);
}
void
Loop::wake()
{
SetEvent(impl_->wake);
}
void
Loop::run()
{
while (true) {
vector<HANDLE> handles{impl_->wake};
vector<pair<uint64_t, bool>> keys{{0, false}};
for (const auto &[id, watch] : impl_->watches) {
if (watch.read) {
handles.push_back(watch.read);
keys.emplace_back(id, false);
}
if (watch.write && watch.write_on) {
handles.push_back(watch.write);
keys.emplace_back(id, true);
}
}
if (handles.size() > MAXIMUM_WAIT_OBJECTS) {
trace("too many IPC watches");
break;
}
const DWORD result = WaitForMultipleObjects(
DWORD(handles.size()), handles.data(), FALSE, INFINITE);
if (result < WAIT_OBJECT_0 || result >= WAIT_OBJECT_0 + handles.size())
break;
const size_t index = result - WAIT_OBJECT_0;
if (!index) {
ResetEvent(impl_->wake);
if (on_wake)
on_wake();
} else if (keys[index].second) {
if (on_write)
on_write(keys[index].first);
} else if (on_read)
on_read(keys[index].first);
}
}
} // namespace dawn::ipc
+40
View File
@@ -0,0 +1,40 @@
//
// ipc-loop.hpp: minimal event loop for IPC services
//
// Copyright The Dawn Authors
// SPDX-License-Identifier: MPL-2.0
//
#pragma once
#include "ipc.hpp"
#include <cstdint>
#include <functional>
#include <memory>
namespace dawn::ipc
{
class Loop
{
struct Impl;
std::unique_ptr<Impl> impl_;
public:
static constexpr uint64_t kListener = UINT64_MAX;
std::function<void(uint64_t)> on_read, on_write;
std::function<void()> on_wake;
Loop();
~Loop();
Loop(const Loop &) = delete;
Loop &operator=(const Loop &) = delete;
[[nodiscard]] bool watch_read(uint64_t id, Waitable w);
void watch_write(uint64_t id, Waitable w, bool enable);
void unwatch(uint64_t id);
void wake();
void run();
};
} // namespace dawn::ipc
+30 -12
View File
@@ -103,20 +103,24 @@ Channel::flush(chrono::milliseconds &budget)
}
bool
Channel::send(span<const byte> payload, chrono::milliseconds &budget)
Channel::send(span<const byte> payload, chrono::milliseconds &budget,
span<const Handle> attachments)
{
if (!conn_.write_payload(payload))
if (!conn_.write_payload(payload, attachments))
return false;
return this->flush(budget);
}
bool
Channel::recv(vector<byte> &payload, chrono::milliseconds &budget)
Channel::recv(vector<byte> &payload, chrono::milliseconds &budget,
vector<Handle> *attachments)
{
while (true) {
switch (conn_.read()) {
case Connection::Status::Frame:
return conn_.take_payload(payload);
if (attachments)
return conn_.take_payload(payload, *attachments);
return conn_.take_plain_payload(payload);
case Connection::Status::NeedMore:
if (!this->wait(Connection::Direction::Read, budget))
return false;
@@ -174,8 +178,12 @@ ServerCore::poll_listen()
return;
trace("accepted %llu from pid %lu", (unsigned long long) id,
(unsigned long) it->second.conn.peer_pid());
if (cfg_.watch_read)
cfg_.watch_read(id, it->second.conn.read_waitable());
if (cfg_.watch_read &&
!cfg_.watch_read(id, it->second.conn.read_waitable())) {
trace("refused connection %llu: event loop is full",
(unsigned long long) id);
this->drop(id);
}
}
}
@@ -203,18 +211,27 @@ ServerCore::poll_read(uint64_t id)
}
vector<byte> payload;
if (!c.conn.take_payload(payload)) {
vector<Handle> attachments;
if (!c.conn.take_payload(payload, attachments)) {
this->drop(id);
return;
}
if (cfg_.on_payload && !cfg_.on_payload(id, payload)) {
this->drop(id);
return;
if (cfg_.on_payload) {
if (!cfg_.on_payload(id, payload, std::move(attachments))) {
this->drop(id);
return;
}
} else {
OwnedHandles owned(std::move(attachments));
(void) owned;
}
const auto after = conns_.find(id);
if (after == conns_.end())
return;
if (cfg_.watch_write)
cfg_.watch_write(id, after->second.conn.write_waitable(),
after->second.conn.wants_write());
if (after->second.closing) {
if (!after->second.conn.wants_write())
this->drop(id);
@@ -247,14 +264,15 @@ ServerCore::poll_write(uint64_t id)
}
bool
ServerCore::send(uint64_t id, span<const byte> payload)
ServerCore::send(
uint64_t id, span<const byte> payload, span<const Handle> attachments)
{
const auto it = conns_.find(id);
if (it == conns_.end())
return false;
Conn &c = it->second;
if (!c.conn.write_payload(payload)) {
if (!c.conn.write_payload(payload, attachments)) {
this->drop(id);
return false;
}
+10 -7
View File
@@ -60,10 +60,10 @@ public:
void set_max_payload(uint32_t limit);
// Both consume budget as time passes, and fail once it runs out.
bool send(
std::span<const std::byte> payload, std::chrono::milliseconds &budget);
bool recv(
std::vector<std::byte> &payload, std::chrono::milliseconds &budget);
bool send(std::span<const std::byte> payload,
std::chrono::milliseconds &budget, std::span<const Handle> attachments);
bool recv(std::vector<std::byte> &payload,
std::chrono::milliseconds &budget, std::vector<Handle> *attachments);
private:
Connection conn_;
@@ -83,12 +83,14 @@ public:
uint32_t max_payload_size = Connection::kMaxPayload;
// One whole payload arrived. Returning false is a protocol
// error and drops the connection.
std::function<bool(uint64_t id, std::span<const std::byte>)> on_payload;
std::function<bool(
uint64_t id, std::span<const std::byte>, std::vector<Handle>)>
on_payload;
// The connection is gone; nothing can be sent on it any more.
std::function<void(uint64_t id)> on_closed;
// Event-loop hooks. Connections are named by an opaque id;
// the Waitable is what the loop has to watch for it.
std::function<void(uint64_t id, Waitable w)> watch_read;
std::function<bool(uint64_t id, Waitable w)> watch_read;
std::function<void(uint64_t id)> unwatch;
std::function<void(uint64_t id, Waitable w, bool enable)> watch_write;
};
@@ -109,7 +111,8 @@ public:
// Queue one payload. False means the connection is gone or refused
// it, in which case it has already been dropped.
bool send(uint64_t id, std::span<const std::byte> payload);
bool send(uint64_t id, std::span<const std::byte> payload,
std::span<const Handle> attachments);
// Write out what is queued, then drop. No further payload is read.
void close_after_flush(uint64_t id);
void drop(uint64_t id);
+125
View File
@@ -0,0 +1,125 @@
//
// ipc-shm-unix.cpp: POSIX anonymous shared pixel storage
//
// Copyright The Dawn Authors
// SPDX-License-Identifier: MPL-2.0
//
#include "ipc-shm.hpp"
#include <cerrno>
#include <cstring>
#include <fcntl.h>
#include <sys/mman.h>
#include <sys/stat.h>
#include <unistd.h>
#if defined __linux__
#include <linux/memfd.h>
#include <sys/syscall.h>
#endif
#include <cstdlib>
#include <string>
using namespace std;
namespace dawn::ipc
{
static int
create_fd()
{
#if defined __linux__
return int(syscall(
SYS_memfd_create, "dawn-pixels", MFD_CLOEXEC | MFD_ALLOW_SEALING));
#elif defined SHM_ANON
return shm_open(SHM_ANON, O_RDWR | O_CLOEXEC, 0600);
#else
for (unsigned i = 0; i < 100; i++) {
const string name = "/dawn-" + to_string(getuid()) + "-" +
to_string(unsigned(arc4random()));
const int fd = shm_open(name.c_str(), O_CREAT | O_EXCL | O_RDWR, 0600);
if (fd >= 0) {
shm_unlink(name.c_str());
fcntl(fd, F_SETFD, FD_CLOEXEC);
return fd;
}
if (errno != EEXIST)
break;
}
return -1;
#endif
}
SharedMemory
SharedMemory::copy(const void *data, size_t size)
{
SharedMemory out;
if (!size)
return out;
const int fd = create_fd();
if (fd < 0 || ftruncate(fd, off_t(size)) != 0) {
if (fd >= 0)
::close(fd);
return out;
}
void *p = mmap(nullptr, size, PROT_READ | PROT_WRITE, MAP_SHARED, fd, 0);
if (p == MAP_FAILED) {
::close(fd);
return out;
}
memcpy(p, data, size);
out.handle_ = fd;
out.data_ = p;
out.size_ = size;
#if defined __linux__
munmap(out.data_, out.size_);
out.data_ = nullptr;
if (fcntl(fd, F_ADD_SEALS, F_SEAL_SHRINK | F_SEAL_WRITE | F_SEAL_SEAL) != 0)
return out;
out.data_ = mmap(nullptr, size, PROT_READ, MAP_SHARED, fd, 0);
if (out.data_ == MAP_FAILED)
out.data_ = nullptr;
#endif
return out;
}
SharedMemory
SharedMemory::map(Handle handle, size_t size)
{
SharedMemory out;
if (handle < 0 || !size) {
close_handle(handle);
return out;
}
struct stat st{};
if (fstat(int(handle), &st) != 0 || st.st_size < 0 ||
uint64_t(st.st_size) < size) {
close_handle(handle);
return out;
}
void *p = mmap(nullptr, size, PROT_READ, MAP_SHARED, int(handle), 0);
if (p == MAP_FAILED) {
close_handle(handle);
return out;
}
out.handle_ = handle;
out.data_ = p;
out.size_ = size;
return out;
}
void
SharedMemory::close()
{
if (data_)
munmap(data_, size_);
close_handle(handle_);
handle_ = kInvalidHandle;
data_ = nullptr;
size_ = 0;
}
} // namespace dawn::ipc
+66
View File
@@ -0,0 +1,66 @@
//
// ipc-shm-windows.cpp: Windows anonymous shared pixel storage
//
// Copyright The Dawn Authors
// SPDX-License-Identifier: MPL-2.0
//
#include "ipc-shm.hpp"
#define WIN32_LEAN_AND_MEAN
#include <windows.h>
#include <cstring>
using namespace std;
namespace dawn::ipc
{
SharedMemory
SharedMemory::copy(const void *data, size_t size)
{
SharedMemory out;
const HANDLE h = CreateFileMappingW(INVALID_HANDLE_VALUE, nullptr,
PAGE_READWRITE, DWORD(uint64_t(size) >> 32), DWORD(size), nullptr);
if (!h)
return out;
void *p = MapViewOfFile(h, FILE_MAP_READ | FILE_MAP_WRITE, 0, 0, size);
if (!p) {
CloseHandle(h);
return out;
}
memcpy(p, data, size);
out.handle_ = (Handle) h;
out.data_ = p;
out.size_ = size;
return out;
}
SharedMemory
SharedMemory::map(Handle handle, size_t size)
{
SharedMemory out;
void *p = MapViewOfFile((HANDLE) handle, FILE_MAP_READ, 0, 0, size);
if (!p) {
close_handle(handle);
return out;
}
out.handle_ = handle;
out.data_ = p;
out.size_ = size;
return out;
}
void
SharedMemory::close()
{
if (data_)
UnmapViewOfFile(data_);
close_handle(handle_);
handle_ = kInvalidHandle;
data_ = nullptr;
size_ = 0;
}
} // namespace dawn::ipc
+44
View File
@@ -0,0 +1,44 @@
//
// ipc-shm.hpp: anonymous shared pixel storage
//
// Copyright The Dawn Authors
// SPDX-License-Identifier: MPL-2.0
//
#pragma once
#include "ipc.hpp"
#include <cstddef>
#include <cstdint>
namespace dawn::ipc
{
class SharedMemory
{
Handle handle_ = kInvalidHandle;
void *data_ = nullptr;
size_t size_ = 0;
public:
SharedMemory() = default;
~SharedMemory();
SharedMemory(SharedMemory &&) noexcept;
SharedMemory &operator=(SharedMemory &&) noexcept;
SharedMemory(const SharedMemory &) = delete;
SharedMemory &operator=(const SharedMemory &) = delete;
static SharedMemory copy(const void *data, size_t size);
static SharedMemory map(Handle handle, size_t size);
[[nodiscard]] bool ok() const { return data_ != nullptr; }
[[nodiscard]] Handle handle() const { return handle_; }
[[nodiscard]] size_t size() const { return size_; }
[[nodiscard]] const uint8_t *data() const
{
return (const uint8_t *) data_;
}
void close();
};
} // namespace dawn::ipc
+182 -7
View File
@@ -9,6 +9,7 @@
#include <cerrno>
#include <cstddef>
#include <cstdlib>
#include <cstring>
#include <fcntl.h>
#include <poll.h>
@@ -19,6 +20,7 @@
#include <string>
#include <utility>
#include <vector>
using namespace std;
@@ -27,12 +29,34 @@ namespace dawn
namespace ipc
{
#ifdef __APPLE__
constexpr int kRecvFlags = 0;
constexpr int kSendFlags = 0;
#else
constexpr int kRecvFlags = MSG_CMSG_CLOEXEC;
constexpr int kSendFlags = MSG_NOSIGNAL;
#endif
void
close_handle(Handle handle)
{
if (handle >= 0)
::close(int(handle));
}
// Abstract name: "\0dawn-<uid>-<service>".
static string
endpoint_name(string_view service)
{
string n;
#ifdef __APPLE__
const char *tmp = getenv("TMPDIR");
n = tmp && *tmp ? tmp : "/tmp/";
if (n.back() != '/')
n.push_back('/');
#else
n.push_back('\0');
#endif
n += "dawn-";
n += to_string(::getuid());
n += '-';
@@ -44,12 +68,17 @@ static bool
fill_addr(string_view service, sockaddr_un &addr, socklen_t &len)
{
const string n = endpoint_name(service);
if (n.empty() || n.size() > sizeof addr.sun_path)
if (n.empty() || n.size() > sizeof addr.sun_path ||
(n.front() != '\0' && n.size() == sizeof addr.sun_path))
return false;
addr = {};
addr.sun_family = AF_UNIX;
memcpy(addr.sun_path, n.data(), n.size());
if (n.front() != '\0')
addr.sun_path[n.size()] = '\0';
len = socklen_t(offsetof(sockaddr_un, sun_path) + n.size());
if (n.front() != '\0')
len++;
return true;
}
@@ -75,6 +104,13 @@ unix_socket(int extra_fl)
::close(fd);
return -1;
}
#ifdef __APPLE__
const int one = 1;
if (::setsockopt(fd, SOL_SOCKET, SO_NOSIGPIPE, &one, sizeof one) != 0) {
::close(fd);
return -1;
}
#endif
return fd;
}
@@ -82,6 +118,18 @@ unix_socket(int extra_fl)
static bool
peer_ok(int fd, uint32_t &pid)
{
#ifdef __APPLE__
xucred cred{};
socklen_t n = sizeof cred;
if (::getsockopt(fd, SOL_LOCAL, LOCAL_PEERCRED, &cred, &n) != 0 ||
n != sizeof cred || cred.cr_uid != ::getuid())
return false;
pid_t peer = 0;
n = sizeof peer;
if (::getsockopt(fd, SOL_LOCAL, LOCAL_PEERPID, &peer, &n) != 0)
return false;
pid = uint32_t(peer);
#else
struct {
pid_t pid;
uid_t uid;
@@ -92,6 +140,7 @@ peer_ok(int fd, uint32_t &pid)
n != sizeof cred || cred.uid != ::getuid())
return false;
pid = uint32_t(cred.pid);
#endif
return true;
}
@@ -104,12 +153,17 @@ struct Connection::Impl {
uint32_t peer_pid = 0;
FrameReader reader;
FrameWriter writer;
vector<Handle> received;
~Impl();
};
Connection::Impl::~Impl()
{
for (Handle h : received)
::close(int(h));
for (Handle h : writer.take_all_attachments())
::close(int(h));
if (fd >= 0)
::close(fd);
}
@@ -181,6 +235,11 @@ Connection::close()
impl_->fd = -1;
}
impl_->ok = false;
for (Handle h : impl_->received)
::close(int(h));
impl_->received.clear();
for (Handle h : impl_->writer.take_all_attachments())
::close(int(h));
impl_->writer.clear();
}
@@ -195,7 +254,14 @@ Connection::read()
while (true) {
const span<byte> buf = m.reader.buffer();
const ssize_t n = ::read(m.fd, buf.data(), buf.size());
iovec iov{buf.data(), buf.size()};
alignas(cmsghdr) char control[CMSG_SPACE(sizeof(int) * 63)]{};
msghdr msg{};
msg.msg_iov = &iov;
msg.msg_iovlen = 1;
msg.msg_control = control;
msg.msg_controllen = sizeof control;
const ssize_t n = ::recvmsg(m.fd, &msg, kRecvFlags);
if (n < 0) {
if (errno == EINTR)
continue;
@@ -210,6 +276,26 @@ Connection::read()
close();
return clean ? Status::Eof : Status::Error;
}
const bool truncated = msg.msg_flags & MSG_CTRUNC;
for (cmsghdr *c = CMSG_FIRSTHDR(&msg); c; c = CMSG_NXTHDR(&msg, c)) {
if (c->cmsg_level != SOL_SOCKET || c->cmsg_type != SCM_RIGHTS ||
c->cmsg_len < CMSG_LEN(0))
continue;
const size_t count = (c->cmsg_len - CMSG_LEN(0)) / sizeof(int);
const auto *fds = reinterpret_cast<const int *>(CMSG_DATA(c));
for (size_t i = 0; i < count; i++) {
if (truncated)
::close(fds[i]);
else {
(void) set_flags(fds[i], 0);
m.received.push_back(fds[i]);
}
}
}
if (truncated) {
close();
return Status::Error;
}
switch (m.reader.advance(size_t(n))) {
case FrameReader::Status::NeedMore:
break;
@@ -223,19 +309,54 @@ Connection::read()
}
bool
Connection::take_payload(vector<byte> &out)
Connection::take_plain_payload(vector<byte> &out)
{
return impl_->reader.take_payload(out);
vector<Handle> attachments;
if (!take_payload(out, attachments))
return false;
for (Handle h : attachments)
::close(int(h));
return true;
}
bool
Connection::write_payload(span<const byte> payload)
Connection::take_payload(vector<byte> &out, vector<Handle> &attachments)
{
Impl &m = *impl_;
const size_t count = m.reader.attachment_count();
if (m.received.size() != count) {
close();
return false;
}
if (!m.reader.take_plain_payload(out))
return false;
attachments.assign(
m.received.begin(), m.received.begin() + ptrdiff_t(count));
m.received.erase(m.received.begin(), m.received.begin() + ptrdiff_t(count));
return true;
}
bool
Connection::write_payload(
span<const byte> payload, span<const Handle> attachments)
{
Impl &m = *impl_;
if (!m.ok)
return false;
if (m.writer.push(payload))
vector<Handle> owned;
for (Handle handle : attachments) {
const int copy = fcntl(int(handle), F_DUPFD_CLOEXEC, 0);
if (copy < 0) {
for (Handle h : owned)
::close(int(h));
return false;
}
owned.push_back(copy);
}
if (m.writer.push(payload, owned))
return true;
for (Handle h : owned)
::close(int(h));
close();
return false;
}
@@ -249,7 +370,24 @@ Connection::flush()
while (!m.writer.empty()) {
const span<const byte> p = m.writer.pending();
const ssize_t n = ::write(m.fd, p.data(), p.size());
const span<const Handle> attachments = m.writer.pending_attachments();
iovec iov{const_cast<byte *>(p.data()), p.size()};
alignas(cmsghdr) char control[CMSG_SPACE(sizeof(int) * 63)]{};
msghdr msg{};
msg.msg_iov = &iov;
msg.msg_iovlen = 1;
if (!attachments.empty()) {
msg.msg_control = control;
msg.msg_controllen = CMSG_SPACE(sizeof(int) * attachments.size());
cmsghdr *c = CMSG_FIRSTHDR(&msg);
c->cmsg_level = SOL_SOCKET;
c->cmsg_type = SCM_RIGHTS;
c->cmsg_len = CMSG_LEN(sizeof(int) * attachments.size());
auto *fds = reinterpret_cast<int *>(CMSG_DATA(c));
for (size_t i = 0; i < attachments.size(); i++)
fds[i] = int(attachments[i]);
}
const ssize_t n = ::sendmsg(m.fd, &msg, kSendFlags);
if (n < 0) {
if (errno == EINTR)
continue;
@@ -262,6 +400,11 @@ Connection::flush()
close();
return false;
}
if (!attachments.empty()) {
for (Handle h : attachments)
::close(int(h));
m.writer.attachments_sent();
}
m.writer.consume(size_t(n));
}
return true;
@@ -302,6 +445,7 @@ Connection::wait(Direction dir, int timeout_ms)
struct Listener::Impl {
int fd = -1;
string path;
~Impl();
};
@@ -310,6 +454,8 @@ Listener::Impl::~Impl()
{
if (fd >= 0)
::close(fd);
if (!path.empty())
::unlink(path.c_str());
}
Listener::Listener() : impl_(make_unique<Impl>())
@@ -340,6 +486,10 @@ Listener::close()
::close(impl_->fd);
impl_->fd = -1;
}
if (!impl_->path.empty()) {
::unlink(impl_->path.c_str());
impl_->path.clear();
}
}
Connection
@@ -361,6 +511,13 @@ Listener::accept()
::close(fd);
continue;
}
#ifdef __APPLE__
const int one = 1;
if (::setsockopt(fd, SOL_SOCKET, SO_NOSIGPIPE, &one, sizeof one) != 0) {
::close(fd);
continue;
}
#endif
Connection conn(fd);
conn.impl_->peer_pid = pid;
@@ -389,6 +546,21 @@ Endpoint::listen(string_view service)
if (errno == EINTR)
continue;
const int e = errno;
trace("bind failed: %s", strerror(e));
#ifdef __APPLE__
if (e == EADDRINUSE) {
const int probe = unix_socket(0);
const int result =
probe < 0 ? -1 : ::connect(probe, (sockaddr *) &addr, len);
const int connect_error = errno;
if (probe >= 0)
::close(probe);
if (result != 0 && connect_error == ECONNREFUSED) {
::unlink(addr.sun_path);
continue;
}
}
#endif
::close(fd);
if (e == EADDRINUSE)
out.status = ListenStatus::InUse;
@@ -405,6 +577,9 @@ Endpoint::listen(string_view service)
}
out.listener.impl_->fd = fd;
#ifdef __APPLE__
out.listener.impl_->path = addr.sun_path;
#endif
out.status = ListenStatus::Ok;
return out;
}
+143 -23
View File
@@ -25,6 +25,13 @@ namespace dawn
namespace ipc
{
void
close_handle(Handle handle)
{
if ((HANDLE) handle && (HANDLE) handle != INVALID_HANDLE_VALUE)
::CloseHandle((HANDLE) handle);
}
// Overlapped writes pin their buffer until completion; keep that bounded
// rather than hand the kernel a whole output queue.
constexpr size_t kWriteChunk = 256 * 1024;
@@ -36,23 +43,28 @@ struct LocalFreeDeleter {
void operator()(void *p) const { ::LocalFree((HLOCAL) p); }
};
struct RemoteHandles {
size_t remaining;
vector<Handle> handles;
};
// Without an acknowledgement, a peer that closes after reading only part of
// an eight-byte handle value cannot name and close it. The handle then lasts
// until that peer exits; complete values are closed by FrameReader::discard().
} // namespace
// The token user of a process, as an owned SID copy.
static vector<std::byte>
process_sid(DWORD pid)
process_sid(HANDLE process)
{
vector<std::byte> out;
const HANDLE proc =
::OpenProcess(PROCESS_QUERY_LIMITED_INFORMATION, FALSE, pid);
if (!proc)
if (!process)
return out;
HANDLE token = nullptr;
if (!::OpenProcessToken(proc, TOKEN_QUERY, &token)) {
::CloseHandle(proc);
if (!::OpenProcessToken(process, TOKEN_QUERY, &token))
return out;
}
DWORD need = 0;
::GetTokenInformation(token, TokenUser, nullptr, 0, &need);
@@ -66,14 +78,26 @@ process_sid(DWORD pid)
out.clear();
}
::CloseHandle(token);
::CloseHandle(proc);
return out;
}
static vector<std::byte>
process_sid_for_pid(DWORD pid)
{
const HANDLE process =
::OpenProcess(PROCESS_QUERY_LIMITED_INFORMATION, FALSE, pid);
if (!process)
return {};
auto sid = process_sid(process);
::CloseHandle(process);
return sid;
}
static const vector<std::byte> &
own_sid()
{
static const vector<std::byte> sid = process_sid(::GetCurrentProcessId());
static const vector<std::byte> sid =
process_sid_for_pid(::GetCurrentProcessId());
return sid;
}
@@ -185,8 +209,21 @@ client_is_own_user(HANDLE pipe)
// --- Connection --------------------------------------------------------------
static void
close_peer_handles(HANDLE process, span<const Handle> handles)
{
for (Handle handle : handles) {
HANDLE local = nullptr;
if (::DuplicateHandle(process, (HANDLE) handle, ::GetCurrentProcess(),
&local, 0, FALSE,
DUPLICATE_CLOSE_SOURCE | DUPLICATE_SAME_ACCESS))
::CloseHandle(local);
}
}
struct Connection::Impl {
HANDLE pipe = INVALID_HANDLE_VALUE;
HANDLE peer_process = nullptr;
bool ok = false;
bool eof = false;
uint32_t peer_pid = 0;
@@ -199,12 +236,38 @@ struct Connection::Impl {
bool write_pending = false;
vector<std::byte> inflight;
FrameWriter writer;
size_t queued_bytes = 0;
vector<RemoteHandles> remote_handles;
~Impl();
void shutdown();
bool set_peer_process(uint32_t pid);
void complete_write(size_t size);
bool post_read();
};
bool
Connection::Impl::set_peer_process(uint32_t pid)
{
peer_process = ::OpenProcess(
PROCESS_DUP_HANDLE | PROCESS_QUERY_LIMITED_INFORMATION, FALSE, pid);
if (!peer_process)
return false;
peer_pid = pid;
return true;
}
void
Connection::Impl::complete_write(size_t size)
{
queued_bytes -= min(size, queued_bytes);
for (auto &pending : remote_handles)
pending.remaining =
pending.remaining > size ? pending.remaining - size : 0;
while (!remote_handles.empty() && remote_handles.front().remaining == 0)
remote_handles.erase(remote_handles.begin());
}
void
Connection::Impl::shutdown()
{
@@ -213,13 +276,22 @@ Connection::Impl::shutdown()
// cancellation completes, so wait for it before anything goes away.
::CancelIoEx(pipe, nullptr);
DWORD n = 0;
if (read_pending)
::GetOverlappedResult(pipe, &rov, &n, TRUE);
if (write_pending)
::GetOverlappedResult(pipe, &wov, &n, TRUE);
if (read_pending && ::GetOverlappedResult(pipe, &rov, &n, TRUE) && n)
(void) reader.advance(n);
read_pending = false;
if (write_pending && ::GetOverlappedResult(pipe, &wov, &n, TRUE))
complete_write(n);
::CloseHandle(pipe);
pipe = INVALID_HANDLE_VALUE;
}
reader.discard();
for (const auto &pending : remote_handles)
close_peer_handles(peer_process, pending.handles);
remote_handles.clear();
queued_bytes = 0;
if (peer_process)
::CloseHandle(peer_process);
peer_process = nullptr;
read_pending = false;
write_pending = false;
if (rov.hEvent)
@@ -262,6 +334,8 @@ Connection::Connection(Handle h) : impl_(make_unique<Impl>())
return;
impl_->pipe = pipe;
impl_->reader.set_inline_handles(true);
impl_->writer.set_inline_handles(true);
impl_->rov.hEvent = ::CreateEventW(nullptr, TRUE, FALSE, nullptr);
impl_->wov.hEvent = ::CreateEventW(nullptr, TRUE, FALSE, nullptr);
if (!impl_->rov.hEvent || !impl_->wov.hEvent) {
@@ -380,19 +454,47 @@ Connection::read()
}
bool
Connection::take_payload(vector<std::byte> &out)
Connection::take_plain_payload(vector<std::byte> &out)
{
return impl_->reader.take_payload(out);
vector<Handle> attachments;
if (!take_payload(out, attachments))
return false;
for (Handle h : attachments)
close_handle(h);
return true;
}
bool
Connection::write_payload(span<const std::byte> payload)
Connection::take_payload(vector<std::byte> &out, vector<Handle> &attachments)
{
return impl_->reader.take_payload(out, attachments);
}
bool
Connection::write_payload(
span<const std::byte> payload, span<const Handle> attachments)
{
Impl &m = *impl_;
if (!m.ok)
return false;
if (m.writer.push(payload))
vector<Handle> duplicated;
duplicated.reserve(attachments.size());
for (Handle h : attachments) {
HANDLE copy = nullptr;
if (!::DuplicateHandle(::GetCurrentProcess(), (HANDLE) h,
m.peer_process, &copy, FILE_MAP_READ, FALSE, 0)) {
close_peer_handles(m.peer_process, duplicated);
return false;
}
duplicated.push_back((Handle) copy);
}
if (m.writer.push(payload, duplicated)) {
m.queued_bytes += 4 + duplicated.size() * 8 + payload.size();
if (!duplicated.empty())
m.remote_handles.push_back({m.queued_bytes, std::move(duplicated)});
return true;
}
close_peer_handles(m.peer_process, duplicated);
close();
return false;
}
@@ -418,6 +520,7 @@ Connection::flush()
close();
return false;
}
m.complete_write(n);
m.inflight.erase(m.inflight.begin(),
m.inflight.begin() + min(size_t(n), m.inflight.size()));
}
@@ -609,11 +712,19 @@ Listener::accept()
}
ULONG client_pid = 0;
::GetNamedPipeClientProcessId(pipe, &client_pid);
if (!::GetNamedPipeClientProcessId(pipe, &client_pid)) {
trace("cannot identify client: %lu", ::GetLastError());
::DisconnectNamedPipe(pipe);
::CloseHandle(pipe);
return {};
}
Connection conn((Handle) pipe);
if (conn.ok())
conn.impl_->peer_pid = uint32_t(client_pid);
if (conn.ok() && !conn.impl_->set_peer_process(uint32_t(client_pid))) {
trace("cannot retain client process %lu: %lu", client_pid,
::GetLastError());
conn.close();
}
return conn;
}
@@ -678,8 +789,7 @@ Endpoint::connect(string_view service)
}
ULONG server_pid = 0;
if (!::GetNamedPipeServerProcessId(pipe, &server_pid) ||
!sid_is_own(process_sid(server_pid))) {
if (!::GetNamedPipeServerProcessId(pipe, &server_pid)) {
trace("rejected server pid %lu: %lu", server_pid, ::GetLastError());
::CloseHandle(pipe);
return out;
@@ -688,7 +798,17 @@ Endpoint::connect(string_view service)
out.conn = Connection((Handle) pipe);
if (!out.conn.ok())
return out;
out.conn.impl_->peer_pid = uint32_t(server_pid);
if (!out.conn.impl_->set_peer_process(uint32_t(server_pid))) {
trace("cannot retain server process %lu: %lu", server_pid,
::GetLastError());
out.conn.close();
return out;
}
if (!sid_is_own(process_sid(out.conn.impl_->peer_process))) {
trace("rejected server identity %lu", server_pid);
out.conn.close();
return out;
}
out.status = ConnectStatus::Ok;
return out;
}
+144 -10
View File
@@ -6,6 +6,7 @@
//
#include "ipc.hpp"
#include "ipc-shm.hpp"
#include <cstdarg>
#include <cstdio>
@@ -19,6 +20,51 @@ namespace dawn
namespace ipc
{
// --- Ownership ---------------------------------------------------------------
OwnedHandles::OwnedHandles(vector<Handle> handles)
: handles_(std::move(handles))
{
}
OwnedHandles::~OwnedHandles()
{
for (Handle handle : handles_)
close_handle(handle);
}
Handle
OwnedHandles::take(size_t index)
{
if (index >= handles_.size())
return kInvalidHandle;
return exchange(handles_[index], kInvalidHandle);
}
SharedMemory::~SharedMemory()
{
close();
}
SharedMemory::SharedMemory(SharedMemory &&other) noexcept
{
*this = std::move(other);
}
SharedMemory &
SharedMemory::operator=(SharedMemory &&other) noexcept
{
if (this != &other) {
close();
handle_ = exchange(other.handle_, kInvalidHandle);
data_ = exchange(other.data_, nullptr);
size_ = exchange(other.size_, 0);
}
return *this;
}
// --- Codec -------------------------------------------------------------------
Encoder::Encoder(vector<byte> &buf) : buf_(buf)
{
}
@@ -28,7 +74,7 @@ Encoder::put_be(uint64_t v, size_t width)
{
if (!ok_)
return;
for (size_t i = width; i > 0; --i)
for (size_t i = width; i > 0; i--)
buf_.push_back(byte((v >> ((i - 1) * 8)) & 0xff));
}
@@ -377,12 +423,16 @@ FrameReader::advance(size_t n)
if (state_ == State::Length) {
if (got_ < 4)
return Status::NeedMore;
const uint32_t size = (uint32_t(length_[0]) << 24) |
const uint32_t packed = (uint32_t(length_[0]) << 24) |
(uint32_t(length_[1]) << 16) | (uint32_t(length_[2]) << 8) |
uint32_t(length_[3]);
if (packed & 0x80000000u)
return Status::Error;
const uint32_t size = packed & 0x01ffffffu;
attachment_count_ = uint8_t((packed >> 25) & 0x3f);
if (size == 0 || size > limit_)
return Status::Error;
payload_.resize(size);
payload_.resize(size + (inline_handles_ ? 8 * attachment_count_ : 0));
got_ = 0;
state_ = State::Payload;
return Status::NeedMore;
@@ -403,16 +453,56 @@ FrameReader::set_limit(uint32_t limit)
}
bool
FrameReader::take_payload(vector<byte> &out)
FrameReader::take_plain_payload(vector<byte> &out)
{
vector<Handle> ignored;
return take_payload(out, ignored);
}
bool
FrameReader::take_payload(vector<byte> &out, vector<Handle> &handles)
{
if (!have_frame_)
return false;
out = std::move(payload_);
handles.clear();
const size_t prefix = inline_handles_ ? 8 * attachment_count_ : 0;
if (prefix) {
handles.reserve(attachment_count_);
for (size_t i = 0; i < attachment_count_; i++) {
uint64_t v = 0;
for (size_t j = 0; j < 8; j++)
v = (v << 8) | uint8_t(payload_[i * 8 + j]);
handles.push_back(Handle(v));
}
out.assign(payload_.begin() + ptrdiff_t(prefix), payload_.end());
} else {
out = std::move(payload_);
}
payload_.clear();
have_frame_ = false;
attachment_count_ = 0;
return true;
}
void
FrameReader::discard()
{
const size_t available = have_frame_ ? payload_.size() : got_;
const size_t count =
inline_handles_ ? min<size_t>(attachment_count_, available / 8) : 0;
for (size_t i = 0; i < count; i++) {
uint64_t value = 0;
for (size_t j = 0; j < 8; j++)
value = (value << 8) | uint8_t(payload_[i * 8 + j]);
close_handle(Handle(value));
}
payload_.clear();
state_ = State::Length;
have_frame_ = false;
got_ = 0;
attachment_count_ = 0;
}
bool
FrameWriter::empty() const
{
@@ -424,7 +514,13 @@ FrameWriter::pending() const
{
if (empty())
return {};
return {q_.data() + off_, q_.size() - off_};
size_t end = q_.size();
for (const auto &a : attachments_)
if (a.offset > off_) {
end = a.offset;
break;
}
return {q_.data() + off_, end - off_};
}
void
@@ -436,27 +532,63 @@ FrameWriter::set_limit(uint32_t limit)
}
bool
FrameWriter::push(span<const byte> payload)
FrameWriter::push(span<const byte> payload, span<const Handle> attachments)
{
if (payload.empty() || payload.size() > limit_)
if (payload.empty() || payload.size() > limit_ || attachments.size() > 63)
return false;
if (off_ > 0) {
q_.erase(q_.begin(), q_.begin() + ptrdiff_t(off_));
for (auto &a : attachments_)
a.offset -= off_;
off_ = 0;
}
const size_t add = 4 + payload.size();
const size_t handles_size = inline_handles_ ? attachments.size() * 8 : 0;
const size_t add = 4 + handles_size + payload.size();
if (add > kMaxQueue || q_.size() > kMaxQueue - add)
return false;
const uint32_t n = uint32_t(payload.size());
const uint32_t n =
uint32_t(payload.size()) | (uint32_t(attachments.size()) << 25);
if (!attachments.empty() && !inline_handles_)
attachments_.push_back(
{q_.size(), {attachments.begin(), attachments.end()}});
q_.push_back(byte((n >> 24) & 0xff));
q_.push_back(byte((n >> 16) & 0xff));
q_.push_back(byte((n >> 8) & 0xff));
q_.push_back(byte(n & 0xff));
if (inline_handles_)
for (Handle h : attachments)
for (size_t i = 8; i > 0; i--)
q_.push_back(byte((uint64_t(h) >> ((i - 1) * 8)) & 0xff));
q_.insert(q_.end(), payload.begin(), payload.end());
return true;
}
span<const Handle>
FrameWriter::pending_attachments() const
{
if (!attachments_.empty() && attachments_.front().offset == off_)
return attachments_.front().handles;
return {};
}
void
FrameWriter::attachments_sent()
{
if (!attachments_.empty() && attachments_.front().offset == off_)
attachments_.erase(attachments_.begin());
}
vector<Handle>
FrameWriter::take_all_attachments()
{
vector<Handle> out;
for (auto &point : attachments_)
out.insert(out.end(), point.handles.begin(), point.handles.end());
attachments_.clear();
return out;
}
void
FrameWriter::consume(size_t n)
{
@@ -465,6 +597,7 @@ FrameWriter::consume(size_t n)
return;
q_.clear();
off_ = 0;
attachments_.clear();
}
void
@@ -472,6 +605,7 @@ FrameWriter::clear()
{
q_.clear();
off_ = 0;
attachments_.clear();
}
} // namespace ipc
+52 -16
View File
@@ -89,6 +89,30 @@ template <typename TView> struct Received {
TView view;
};
// A file descriptor on POSIX, a HANDLE on Windows. Both compare equal to
// -1 when invalid.
using Handle = std::intptr_t;
inline constexpr Handle kInvalidHandle = -1;
// What an event loop waits on: the connection itself on POSIX, an
// overlapped completion event on Windows.
using Waitable = Handle;
void close_handle(Handle handle);
class OwnedHandles
{
std::vector<Handle> handles_;
public:
explicit OwnedHandles(std::vector<Handle> handles);
~OwnedHandles();
OwnedHandles(const OwnedHandles &) = delete;
OwnedHandles &operator=(const OwnedHandles &) = delete;
[[nodiscard]] size_t size() const { return handles_.size(); }
[[nodiscard]] bool empty() const { return handles_.empty(); }
Handle take(size_t index);
};
// --- Framing -----------------------------------------------------------------
// Two-state frame codec: a big-endian u32 length prefix, then exactly that
@@ -104,6 +128,8 @@ class FrameReader
std::byte length_[4]{};
size_t got_ = 0;
uint32_t limit_ = kMaxPayload;
uint8_t attachment_count_ = 0;
bool inline_handles_ = false;
std::vector<std::byte> payload_;
public:
@@ -120,7 +146,13 @@ public:
[[nodiscard]] bool ready() const;
// True at a frame boundary, the only place a peer may close cleanly.
[[nodiscard]] bool idle() const;
bool take_payload(std::vector<std::byte> &out);
bool take_plain_payload(std::vector<std::byte> &out);
bool take_payload(
std::vector<std::byte> &out, std::vector<Handle> &handles);
// Close complete inline handles and reset even an incomplete frame.
void discard();
[[nodiscard]] uint8_t attachment_count() const { return attachment_count_; }
void set_inline_handles(bool enable) { inline_handles_ = enable; }
// Lower the accepted frame size; values above kMaxPayload are clamped.
// A length prefix is believed before the payload arrives, so this is
// what bounds the allocation an unauthenticated peer can ask for.
@@ -132,38 +164,39 @@ public:
// of whatever it hands to the operating system.
class FrameWriter
{
struct AttachmentPoint {
size_t offset;
std::vector<Handle> handles;
};
std::vector<std::byte> q_;
size_t off_ = 0;
uint32_t limit_ = FrameReader::kMaxPayload;
std::vector<AttachmentPoint> attachments_;
bool inline_handles_ = false;
public:
static constexpr size_t kMaxQueue = 64 * 1024 * 1024;
// Prefix payload with its length and enqueue it.
bool push(std::span<const std::byte> payload);
bool push(std::span<const std::byte> payload,
std::span<const Handle> attachments);
[[nodiscard]] bool empty() const;
[[nodiscard]] std::span<const std::byte> pending() const;
void consume(size_t n);
[[nodiscard]] std::span<const Handle> pending_attachments() const;
void attachments_sent();
std::vector<Handle> take_all_attachments();
void clear();
// Refuse to enqueue frames the peer would refuse to read.
void set_limit(uint32_t limit);
void set_inline_handles(bool enable) { inline_handles_ = enable; }
};
// --- Transport ---------------------------------------------------------------
// Set DN_IPC_DEBUG to have the transport explain itself on stderr.
// Everything here fails by quietly running a second dn instead.
DAWN_FORMAT(1, 2)
void trace(const char *fmt, ...);
// A file descriptor on POSIX, a HANDLE on Windows. Both compare equal to
// -1 when invalid.
using Handle = std::intptr_t;
inline constexpr Handle kInvalidHandle = -1;
// What an event loop waits on: the connection itself on POSIX, an
// overlapped completion event on Windows.
using Waitable = Handle;
DAWN_FORMAT(1, 2) void trace(const char *fmt, ...);
// A framed full-duplex byte stream. POSIX drives it from readiness on a
// nonblocking Unix socket, Windows from completion of overlapped named
@@ -203,13 +236,16 @@ public:
// Diagnostics and window activation only; never authentication.
[[nodiscard]] uint32_t peer_pid() const;
// Harvest input. Frame means take_payload() will succeed.
// Harvest input. Frame means one of the take methods will succeed.
Status read();
bool take_payload(std::vector<std::byte> &out);
bool take_plain_payload(std::vector<std::byte> &out);
bool take_payload(
std::vector<std::byte> &out, std::vector<Handle> &attachments);
// Queue one payload; it is written by flush(), which returns true
// once the queue has drained.
bool write_payload(std::span<const std::byte> payload);
bool write_payload(std::span<const std::byte> payload,
std::span<const Handle> attachments);
bool flush();
// Block until the direction can make progress. For BlockingClient;
+7 -6
View File
@@ -4,7 +4,6 @@ target_link_libraries(gen-fixtures PRIVATE
${DAWN_LCMS2_TARGET}
)
pkg_check_modules(LIBPNG IMPORTED_TARGET libpng)
if (LIBPNG_FOUND)
add_executable(gen-sample gen-sample.c)
target_link_libraries(gen-sample PRIVATE
@@ -129,11 +128,13 @@ add_executable(test-lxdr test-lxdr.cpp)
target_link_libraries(test-lxdr PRIVATE dawn::libdn)
add_test(NAME lxdr COMMAND test-lxdr)
if(NOT APPLE)
add_executable(test-ipc test-ipc.cpp)
target_link_libraries(test-ipc PRIVATE dawn::libdn)
add_test(NAME ipc COMMAND test-ipc)
endif()
add_executable(test-ipc test-ipc.cpp)
target_link_libraries(test-ipc PRIVATE dawn::libdn)
add_test(NAME ipc COMMAND test-ipc)
add_executable(test-shm test-shm.cpp)
target_link_libraries(test-shm PRIVATE dawn::libdn)
add_test(NAME shm COMMAND test-shm)
# The main desktop file is rewritten by dn-update-desktop-files on install.
find_program(DAWN_DESKTOP_FILE_VALIDATE desktop-file-validate)
+393 -12
View File
@@ -6,9 +6,12 @@
//
#include "libdn/ipc-instance.hpp"
#include "libdn/ipc-loop.hpp"
#include "libdn/ipc-shm.hpp"
#include "libdn/ipc.hpp"
#include "test.hpp"
#include <algorithm>
#include <cerrno>
#include <cstddef>
#include <cstdint>
@@ -24,8 +27,12 @@
#include <vector>
#ifndef _WIN32
#include <fcntl.h>
#include <sys/socket.h>
#include <unistd.h>
#else
#define WIN32_LEAN_AND_MEAN
#include <windows.h>
#endif
using namespace std;
@@ -58,7 +65,7 @@ struct SocketPair {
static bool
write_all(int fd, const void *p, size_t n)
{
const auto *b = static_cast<const uint8_t *>(p);
const auto *b = (const uint8_t *) p;
size_t off = 0;
while (off < n) {
const ssize_t w = ::write(fd, b + off, n - off);
@@ -74,6 +81,24 @@ write_all(int fd, const void *p, size_t n)
return true;
}
static bool
send_fd_with_bytes(int socket, int fd, span<const uint8_t> bytes)
{
iovec iov{const_cast<uint8_t *>(bytes.data()), bytes.size()};
alignas(cmsghdr) char control[CMSG_SPACE(sizeof fd)]{};
msghdr message{};
message.msg_iov = &iov;
message.msg_iovlen = 1;
message.msg_control = control;
message.msg_controllen = sizeof control;
cmsghdr *header = CMSG_FIRSTHDR(&message);
header->cmsg_level = SOL_SOCKET;
header->cmsg_type = SCM_RIGHTS;
header->cmsg_len = CMSG_LEN(sizeof fd);
memcpy(CMSG_DATA(header), &fd, sizeof fd);
return sendmsg(socket, &message, 0) == ssize_t(bytes.size());
}
static void
put_u32be(uint8_t *out, uint32_t n)
{
@@ -123,7 +148,7 @@ test_fragmented()
}
vector<byte> got;
CHECK(conn.take_payload(got));
CHECK(conn.take_plain_payload(got));
CHECK(payload_eq(got, kPayload));
}
@@ -148,12 +173,12 @@ test_two_frames_one_write()
CHECK(conn.read() == dawn::ipc::Connection::Status::Frame);
vector<byte> first;
CHECK(conn.take_payload(first));
CHECK(conn.take_plain_payload(first));
CHECK(payload_eq(first, kA));
CHECK(conn.read() == dawn::ipc::Connection::Status::Frame);
vector<byte> second;
CHECK(conn.take_payload(second));
CHECK(conn.take_plain_payload(second));
CHECK(payload_eq(second, kB));
}
@@ -202,13 +227,239 @@ test_write_read_pair()
static constexpr uint8_t kPayload[] = {1, 2, 3, 4, 5};
const auto payload = as_bytes(span(kPayload));
CHECK(a.write_payload(payload));
CHECK(a.write_payload(payload, {}));
CHECK(a.flush());
CHECK(b.read() == dawn::ipc::Connection::Status::Frame);
vector<byte> got;
CHECK(b.take_payload(got));
CHECK(b.take_plain_payload(got));
CHECK(payload_eq(got, kPayload));
}
static void
test_attachment_pair()
{
SocketPair pair;
dawn::ipc::Connection a(pair.take(0)), b(pair.take(1));
static constexpr char kShared[] = "dawn";
auto memory = dawn::ipc::SharedMemory::copy(kShared, 4);
CHECK(memory.ok());
static constexpr byte payload[]{byte{42}};
const dawn::ipc::Handle handle = memory.handle();
CHECK(a.write_payload(payload, span(&handle, 1)));
CHECK(a.flush());
CHECK(b.read() == dawn::ipc::Connection::Status::Frame);
vector<byte> got;
vector<dawn::ipc::Handle> handles;
CHECK(b.take_payload(got, handles));
CHECK(got.size() == 1 && got[0] == byte{42});
CHECK(handles.size() == 1);
if (handles.size() == 1) {
auto mapped = dawn::ipc::SharedMemory::map(handles[0], 4);
CHECK(mapped.ok() && memcmp(mapped.data(), "dawn", 4) == 0);
}
}
static vector<int>
open_fds()
{
vector<int> out;
for (int fd = 0; fd < 256; fd++)
if (fcntl(fd, F_GETFD) >= 0)
out.push_back(fd);
return out;
}
static void
test_close_discards_attachments()
{
SocketPair pair;
dawn::ipc::Connection a(pair.take(0)), b(pair.take(1));
const byte data[4]{};
auto memory = dawn::ipc::SharedMemory::copy(data, sizeof data);
const dawn::ipc::Handle handle = memory.handle();
static constexpr byte payload[]{byte{1}};
CHECK(a.write_payload(payload, span(&handle, 1)));
CHECK(a.flush());
const vector<int> before = open_fds();
CHECK(b.read() == dawn::ipc::Connection::Status::Frame);
const vector<int> after = open_fds();
vector<int> received;
for (int fd : after)
if (find(before.begin(), before.end(), fd) == before.end())
received.push_back(fd);
CHECK(received.size() == 1);
b.close();
if (received.size() == 1)
CHECK(fcntl(received[0], F_GETFD) < 0 && errno == EBADF);
}
static void
test_rejects_undeclared_attachment()
{
SocketPair pair;
dawn::ipc::Connection receiver(pair.take(0));
const byte data[4]{};
auto memory = dawn::ipc::SharedMemory::copy(data, sizeof data);
const uint8_t wire[]{0, 0, 0, 1, 42};
CHECK(send_fd_with_bytes(pair.fds[1], int(memory.handle()), wire));
CHECK(receiver.read() == dawn::ipc::Connection::Status::Frame);
vector<byte> payload;
vector<dawn::ipc::Handle> handles;
CHECK(!receiver.take_payload(payload, handles));
CHECK(!receiver.ok());
CHECK(handles.empty());
}
#endif
static void
test_attachment_length_prefix()
{
dawn::ipc::FrameWriter writer;
static constexpr byte payload[]{byte{1}, byte{2}};
CHECK(writer.push(payload, {}));
auto plain = writer.pending();
CHECK(plain.size() == 6);
CHECK(plain[0] == byte{0} && plain[1] == byte{0} && plain[2] == byte{0} &&
plain[3] == byte{2});
writer.clear();
const dawn::ipc::Handle handles[]{3, 4};
CHECK(writer.push(payload, handles));
auto attached = writer.pending();
CHECK(attached.size() == 6);
CHECK(attached[0] == byte{4} && attached[3] == byte{2});
}
static void
test_attachment_reader()
{
dawn::ipc::FrameReader reader;
uint8_t wire[] = {6, 0, 0, 2, 9, 8};
memcpy(reader.buffer().data(), wire, 4);
CHECK(reader.advance(4) == dawn::ipc::FrameReader::Status::NeedMore);
CHECK(reader.attachment_count() == 3);
memcpy(reader.buffer().data(), wire + 4, 2);
CHECK(reader.advance(2) == dawn::ipc::FrameReader::Status::Frame);
vector<byte> payload;
vector<dawn::ipc::Handle> handles{99};
CHECK(reader.take_payload(payload, handles));
CHECK(handles.empty());
CHECK(payload.size() == 2);
}
static void
test_attachment_boundaries()
{
dawn::ipc::FrameWriter writer;
static constexpr byte payload[]{byte{1}};
const dawn::ipc::Handle first[]{10};
const dawn::ipc::Handle second[]{20, 21};
CHECK(writer.push(payload, first));
CHECK(writer.push(payload, {}));
CHECK(writer.push(payload, second));
CHECK(writer.pending_attachments().size() == 1);
CHECK(writer.pending().size() == 10);
writer.attachments_sent();
writer.consume(10);
CHECK(writer.pending_attachments().size() == 2);
CHECK(writer.pending().size() == 5);
(void) writer.take_all_attachments();
}
static void
test_clear_discards_attachment_points()
{
dawn::ipc::FrameWriter writer;
static constexpr byte attached[]{byte{9}};
const dawn::ipc::Handle handle[]{10};
CHECK(writer.push(attached, handle));
CHECK(writer.pending_attachments().size() == 1);
writer.clear();
static constexpr byte plain[]{byte{1}, byte{2}};
CHECK(writer.push(plain, {}));
CHECK(writer.pending_attachments().empty());
const auto pending = writer.pending();
CHECK(pending.size() == 6 && pending[0] == byte{0} &&
pending[1] == byte{0} && pending[2] == byte{0} &&
pending[3] == byte{2} && pending[4] == byte{1} &&
pending[5] == byte{2});
}
static void
test_inline_attachment_encoding()
{
dawn::ipc::FrameWriter writer;
writer.set_inline_handles(true);
const byte payload[]{byte{7}, byte{8}};
const dawn::ipc::Handle sent[]{0x01020304};
CHECK(writer.push(payload, sent));
const auto application = writer.pending();
CHECK(application.size() == 14 && application[0] == byte{2} &&
application[1] == byte{0} && application[2] == byte{0} &&
application[3] == byte{2} && application[12] == byte{7} &&
application[13] == byte{8});
dawn::ipc::FrameReader reader;
reader.set_inline_handles(true);
memcpy(reader.buffer().data(), application.data(), 4);
CHECK(reader.advance(4) == dawn::ipc::FrameReader::Status::NeedMore);
memcpy(reader.buffer().data(), application.data() + 4, 10);
CHECK(reader.advance(10) == dawn::ipc::FrameReader::Status::Frame);
vector<byte> received;
vector<dawn::ipc::Handle> handles;
CHECK(reader.take_payload(received, handles));
CHECK(received.size() == 2 && received[0] == byte{7} &&
received[1] == byte{8});
CHECK(handles.size() == 1 && handles[0] == sent[0]);
}
static void
test_reader_rejects_reserved_bit()
{
dawn::ipc::FrameReader reader;
const byte reserved[]{byte{0x80}, byte{0}, byte{0}, byte{0}};
memcpy(reader.buffer().data(), reserved, sizeof reserved);
CHECK(reader.advance(sizeof reserved) ==
dawn::ipc::FrameReader::Status::Error);
}
#ifdef _WIN32
static void
test_inline_attachment_discard()
{
HANDLE handle = ::CreateEventW(nullptr, TRUE, FALSE, nullptr);
CHECK(handle != nullptr);
if (!handle)
return;
dawn::ipc::FrameWriter writer;
writer.set_inline_handles(true);
const dawn::ipc::Handle sent[]{(dawn::ipc::Handle) handle};
const byte payload[]{byte{1}, byte{2}};
CHECK(writer.push(payload, sent));
const auto frame = writer.pending();
dawn::ipc::FrameReader reader;
reader.set_inline_handles(true);
memcpy(reader.buffer().data(), frame.data(), 4);
CHECK(reader.advance(4) == dawn::ipc::FrameReader::Status::NeedMore);
memcpy(reader.buffer().data(), frame.data() + 4, 9);
CHECK(reader.advance(9) == dawn::ipc::FrameReader::Status::NeedMore);
reader.discard();
DWORD flags = 0;
CHECK(!::GetHandleInformation(handle, &flags));
CHECK(::GetLastError() == ERROR_INVALID_HANDLE);
}
static void
test_loop_refuses_excess_connections()
{
dawn::ipc::Loop loop;
CHECK(loop.watch_read(dawn::ipc::Loop::kListener, 1));
for (uint64_t id = 1; id <= 31; id++)
CHECK(loop.watch_read(id, dawn::ipc::Waitable(id + 1)));
CHECK(!loop.watch_read(32, 33));
}
#endif
// --- Endpoint ----------------------------------------------------------------
@@ -255,7 +506,7 @@ test_endpoint_roundtrip()
static constexpr uint8_t kPayload[] = {9, 8, 7};
const auto payload = as_bytes(span(kPayload));
CHECK(connect.conn.write_payload(payload));
CHECK(connect.conn.write_payload(payload, {}));
CHECK(connect.conn.flush());
while (server.read() == dawn::ipc::Connection::Status::NeedMore) {
@@ -264,7 +515,7 @@ test_endpoint_roundtrip()
break;
}
vector<byte> got;
CHECK(server.take_payload(got));
CHECK(server.take_plain_payload(got));
CHECK(payload_eq(got, kPayload));
// The peer identity is what stands in for a same-user check.
@@ -272,6 +523,120 @@ test_endpoint_roundtrip()
CHECK(connect.conn.peer_pid() != 0);
}
#ifdef _WIN32
static dawn::ipc::Connection
accept_connection(dawn::ipc::Listener &listener)
{
for (int i = 0; i < 100; i++) {
auto connection = listener.accept();
if (connection.ok())
return connection;
::WaitForSingleObject((HANDLE) listener.waitable(), 20);
}
return {};
}
static bool
flush_connection(dawn::ipc::Connection &connection)
{
for (int i = 0; i < 100; i++) {
if (connection.flush())
return true;
if (!connection.ok() ||
connection.wait(dawn::ipc::Connection::Direction::Write, 2000) !=
dawn::ipc::Connection::Ready::Ok)
return false;
}
return false;
}
static bool
read_frame(dawn::ipc::Connection &connection)
{
for (int i = 0; i < 100; i++) {
const auto status = connection.read();
if (status == dawn::ipc::Connection::Status::Frame)
return true;
if (status != dawn::ipc::Connection::Status::NeedMore ||
connection.wait(dawn::ipc::Connection::Direction::Read, 2000) !=
dawn::ipc::Connection::Ready::Ok)
return false;
}
return false;
}
static void
test_windows_attachment_protocol()
{
auto endpoint = dawn::ipc::Endpoint::listen("test-windows-attachments");
CHECK(endpoint.status == dawn::ipc::Endpoint::ListenStatus::Ok);
if (endpoint.status != dawn::ipc::Endpoint::ListenStatus::Ok)
return;
auto client = dawn::ipc::Endpoint::connect("test-windows-attachments");
CHECK(client.status == dawn::ipc::Endpoint::ConnectStatus::Ok);
auto server = accept_connection(endpoint.listener);
CHECK(client.conn.ok() && server.ok());
if (!client.conn.ok() || !server.ok())
return;
const uint8_t first_value = 17, second_value = 29;
auto first = dawn::ipc::SharedMemory::copy(&first_value, 1);
auto second = dawn::ipc::SharedMemory::copy(&second_value, 1);
CHECK(first.ok() && second.ok());
const dawn::ipc::Handle first_handle[]{first.handle()};
const dawn::ipc::Handle second_handle[]{second.handle()};
const byte first_payload[]{byte{1}};
const byte second_payload[]{byte{2}};
CHECK(client.conn.write_payload(first_payload, first_handle));
CHECK(client.conn.write_payload(second_payload, second_handle));
first.close();
second.close();
CHECK(flush_connection(client.conn));
for (uint8_t expected : {uint8_t(17), uint8_t(29)}) {
CHECK(read_frame(server));
vector<byte> payload;
vector<dawn::ipc::Handle> handles;
CHECK(server.take_payload(payload, handles));
CHECK(handles.size() == 1);
if (handles.size() == 1) {
auto memory = dawn::ipc::SharedMemory::map(handles[0], 1);
handles.clear();
CHECK(memory.ok() && memory.data()[0] == expected);
}
}
}
static void
test_windows_attachment_teardown()
{
auto endpoint = dawn::ipc::Endpoint::listen("test-windows-handle-close");
CHECK(endpoint.status == dawn::ipc::Endpoint::ListenStatus::Ok);
if (endpoint.status != dawn::ipc::Endpoint::ListenStatus::Ok)
return;
auto client = dawn::ipc::Endpoint::connect("test-windows-handle-close");
auto server = accept_connection(endpoint.listener);
CHECK(client.conn.ok() && server.ok());
const byte value{};
auto memory = dawn::ipc::SharedMemory::copy(&value, 1);
const dawn::ipc::Handle handle[]{memory.handle()};
const byte payload[]{byte{1}};
CHECK(client.conn.write_payload(payload, handle));
CHECK(flush_connection(client.conn));
client.conn.close();
CHECK(read_frame(server));
vector<byte> received;
vector<dawn::ipc::Handle> handles;
CHECK(server.take_payload(received, handles));
CHECK(handles.size() == 1);
if (handles.size() == 1) {
auto mapped = dawn::ipc::SharedMemory::map(handles[0], 1);
handles.clear();
CHECK(mapped.ok());
}
}
#endif
// --- Instance service --------------------------------------------------------
// Everything below drives dawn::ipc::instance::Server through a hand-built
@@ -375,6 +740,7 @@ Fixture::start()
cfg.max_payload_size = kMaxPayload;
cfg.watch_read = [this](uint64_t id, dawn::ipc::Waitable) {
this->conn_id = id;
return true;
};
cfg.on_request = [this](inst::Call call, const inst::RequestView &req) {
this->requests++;
@@ -402,7 +768,7 @@ bool
Fixture::send(const inst::Frame &frame)
{
const vector<byte> buf = frame_bytes(frame);
return this->peer.write_payload(buf) && this->peer.flush();
return this->peer.write_payload(buf, {}) && this->peer.flush();
}
bool
@@ -412,7 +778,7 @@ Fixture::recv(inst::FrameView &view, vector<byte> &storage)
this->poll();
const auto st = this->peer.read();
if (st == dawn::ipc::Connection::Status::Frame) {
if (!this->peer.take_payload(storage))
if (!this->peer.take_plain_payload(storage))
return false;
dawn::ipc::Decoder dec(storage);
return decode(dec, view) && dec.remaining() == 0;
@@ -436,7 +802,7 @@ Fixture::closed()
return true;
if (st == dawn::ipc::Connection::Status::Frame) {
vector<byte> drop;
(void) this->peer.take_payload(drop);
(void) this->peer.take_plain_payload(drop);
continue;
}
this->peer.wait(dawn::ipc::Connection::Direction::Read, 5);
@@ -768,7 +1134,7 @@ test_instance_oversize()
const vector<byte> buf = frame_bytes(frame);
CHECK(buf.size() > kMaxPayload);
CHECK(f.peer.write_payload(buf));
CHECK(f.peer.write_payload(buf, {}));
(void) f.peer.flush();
CHECK(f.closed());
CHECK(f.requests == 0);
@@ -784,6 +1150,21 @@ main()
{"empty payload", test_empty_payload},
{"oversize payload", test_oversize_length},
{"connection pair", test_write_read_pair},
{"attachment pair", test_attachment_pair},
{"close discards attachments", test_close_discards_attachments},
{"reject undeclared attachment", test_rejects_undeclared_attachment},
#endif
{"attachment length prefix", test_attachment_length_prefix},
{"attachment reader", test_attachment_reader},
{"attachment boundaries", test_attachment_boundaries},
{"clear attachment points", test_clear_discards_attachment_points},
{"inline attachment encoding", test_inline_attachment_encoding},
{"reserved bit rejection", test_reader_rejects_reserved_bit},
#ifdef _WIN32
{"inline attachment discard", test_inline_attachment_discard},
{"loop capacity refusal", test_loop_refuses_excess_connections},
{"Windows attachment protocol", test_windows_attachment_protocol},
{"Windows attachment teardown", test_windows_attachment_teardown},
#endif
{"listener arbitration", test_listen_arbitrates},
{"endpoint round trip", test_endpoint_roundtrip},
+94
View File
@@ -7,7 +7,9 @@
#include "libdn/ipc.hpp"
#include "ipc/imaged.lxdr.hpp"
#include "ipc/instance.lxdr.hpp"
#include "ipc/thumbd.lxdr.hpp"
#include "test.hpp"
#include <cstddef>
@@ -21,6 +23,8 @@
using namespace std;
namespace inst = dawn::ipc::instance;
namespace imaged = dawn::ipc::imaged;
namespace thumbd = dawn::ipc::thumbd;
template <typename T>
static vector<byte>
@@ -387,6 +391,94 @@ 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::Decoder hello_decoder(bytes);
imaged::DaemonHelloReplyView hello_view;
CHECK(decode(hello_decoder, hello_view));
const auto *accepted =
get_if<imaged::DaemonHelloReplyAcceptedView>(&hello_view.value);
CHECK(accepted != nullptr);
if (accepted)
CHECK(accepted->limits.max_blob_size == imaged::kImagedMaxBlobSize);
imaged::Pixmap pixmap;
pixmap.width = 2;
pixmap.height = 3;
pixmap.stride = 16;
pixmap.orientation = 6;
pixmap.pixels.value = imaged::BlobShared{48};
imaged::DecodeResponse response{pixmap, "wuffs", {}, true};
bytes = encoded(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);
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::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);
}
static void
test_blob_envelope_sizes()
{
imaged::DecodeRequest decode;
decode.data.value = imaged::BlobInline{};
imaged::Frame request;
request.payload.value = imaged::PayloadRequest{imaged::Request{1, decode}};
const size_t imaged_overhead = encoded(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);
imaged::DecodeResponse decoded;
decoded.pixmap.width = 2;
decoded.pixmap.height = 2;
decoded.pixmap.stride = 16;
decoded.pixmap.pixels.value = imaged::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();
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);
thumbd::ScaleResponse scale;
scale.width = 1;
scale.height = 1;
scale.rgba8.value = thumbd::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();
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);
}
int
main()
{
@@ -401,5 +493,7 @@ main()
{"invalid enum", test_zero_and_unknown_enum},
{"unknown union tag", test_unknown_union_tag},
{"size limits", test_huge_array_count},
{"daemon schemas", test_daemon_schemas},
{"blob envelope sizes", test_blob_envelope_sizes},
});
}
+64
View File
@@ -0,0 +1,64 @@
//
// test-shm.cpp: anonymous shared-memory tests
//
// Copyright The Dawn Authors
// SPDX-License-Identifier: MPL-2.0
//
#include "libdn/ipc-shm.hpp"
#include "test.hpp"
#include <cstring>
#ifdef _WIN32
#define WIN32_LEAN_AND_MEAN
#include <windows.h>
#else
#include <cerrno>
#include <sys/mman.h>
#include <unistd.h>
#endif
using namespace std;
static dawn::ipc::Handle
duplicate_handle(dawn::ipc::Handle handle)
{
#ifdef _WIN32
HANDLE copy = nullptr;
DuplicateHandle(GetCurrentProcess(), (HANDLE) handle, GetCurrentProcess(),
&copy, FILE_MAP_READ, FALSE, 0);
return dawn::ipc::Handle(copy);
#else
return dup(int(handle));
#endif
}
static void
test_shared_memory()
{
char data[4096]{};
memcpy(data, "shared", 6);
auto owner = dawn::ipc::SharedMemory::copy(data, sizeof data);
CHECK(owner.ok());
auto too_large =
dawn::ipc::SharedMemory::map(duplicate_handle(owner.handle()), 8192);
CHECK(!too_large.ok());
#ifdef __linux__
errno = 0;
CHECK(ftruncate(int(owner.handle()), 2048) != 0);
CHECK(errno == EPERM);
void *writable =
mmap(nullptr, 4096, PROT_WRITE, MAP_SHARED, int(owner.handle()), 0);
CHECK(writable == MAP_FAILED);
#endif
auto reader =
dawn::ipc::SharedMemory::map(duplicate_handle(owner.handle()), 4096);
CHECK(reader.ok() && memcmp(reader.data(), "shared", 6) == 0);
}
int
main()
{
return test::run({{"shared memory", test_shared_memory}});
}
+13
View File
@@ -0,0 +1,13 @@
foreach(target dnimage dnthumb)
add_executable(${target}
${target}.cpp
png-io.cpp
)
target_link_libraries(${target} PRIVATE
dawn::libdn
PkgConfig::LIBPNG
)
target_include_directories(${target} PRIVATE
${PROJECT_SOURCE_DIR}/libdn
)
endforeach()
+215
View File
@@ -0,0 +1,215 @@
//
// dnimage.cpp: command-line client for dnimaged
//
// Copyright The Dawn Authors
// SPDX-License-Identifier: MPL-2.0
//
#include "ipc-rpc.hpp"
#include "ipc-shm.hpp"
#include "ipc/imaged.lxdr.hpp"
#include "libdn.h"
#include "png-io.hpp"
#include <chrono>
#include <cstdio>
#include <variant>
#include <vector>
#ifdef _WIN32
#include <fcntl.h>
#include <io.h>
#endif
using namespace std;
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,
chrono::milliseconds &budget)
{
if (!channel.recv(storage, budget, &handles))
return false;
dawn::ipc::Decoder decoder(storage);
return decode(decoder, view) && !decoder.remaining();
}
// --- Main --------------------------------------------------------------------
int
main()
{
#ifdef _WIN32
_setmode(_fileno(stdin), _O_BINARY);
_setmode(_fileno(stdout), _O_BINARY);
#endif
vector<byte> input;
byte chunk[65536];
while (size_t n = fread(chunk, 1, sizeof chunk, stdin)) {
if (uint64_t(input.size()) > uint64_t(proto::kImagedMaxBlobSize) - n) {
fprintf(stderr, "dnimage: input is too large\n");
return 1;
}
input.insert(input.end(), chunk, chunk + n);
}
if (ferror(stdin) || input.empty()) {
fprintf(stderr, "dnimage: cannot read input\n");
return 1;
}
chrono::milliseconds budget = chrono::seconds(10);
auto channel = dawn::ipc::Channel::connect("imaged", budget);
if (!channel.ok()) {
fprintf(stderr, "dnimage: dnimaged is not running\n");
return 1;
}
proto::Frame hello;
hello.payload.value =
proto::PayloadHello{proto::Hello{proto::kImagedProtocolVersion, ""}};
if (!channel.send(encoded(hello), budget, {})) {
fprintf(stderr, "dnimage: handshake failed\n");
return 1;
}
vector<byte> wire;
vector<dawn::ipc::Handle> reply_handles;
proto::FrameView reply;
const bool received = receive(channel, reply, wire, reply_handles, budget);
dawn::ipc::OwnedHandles owned_reply_handles(std::move(reply_handles));
if (!received || !owned_reply_handles.empty() ||
!holds_alternative<proto::PayloadHelloReplyView>(reply.payload.value)) {
fprintf(stderr, "dnimage: incompatible daemon\n");
return 1;
}
const auto *accepted = get_if<proto::DaemonHelloReplyAcceptedView>(
&get<proto::PayloadHelloReplyView>(reply.payload.value)
.hello_reply.value);
if (!accepted || !accepted->limits.max_payload_size ||
accepted->limits.max_payload_size >
dawn::ipc::Connection::kMaxPayload ||
!accepted->limits.max_blob_size ||
accepted->limits.max_blob_size > proto::kImagedMaxBlobSize) {
fprintf(stderr, "dnimage: invalid daemon limits\n");
return 1;
}
const uint32_t max_payload = accepted->limits.max_payload_size;
const uint64_t max_blob = accepted->limits.max_blob_size;
channel.set_max_payload(max_payload);
if (input.size() > max_blob) {
fprintf(stderr, "dnimage: input is too large\n");
return 1;
}
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();
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);
}
if (request_wire.size() > max_payload ||
!channel.send(request_wire, budget, request_handles)) {
fprintf(stderr, "dnimage: request failed\n");
return 1;
}
vector<dawn::ipc::Handle> handles;
proto::FrameView response_frame;
const bool response_received =
receive(channel, response_frame, wire, handles, budget);
dawn::ipc::OwnedHandles owned_handles(std::move(handles));
if (!response_received) {
fprintf(stderr, "dnimage: decode failed\n");
return 1;
}
auto payload =
get_if<proto::PayloadResponseView>(&response_frame.payload.value);
if (!payload)
return 1;
if (payload->response.id != 1) {
fprintf(stderr, "dnimage: response request ID mismatch\n");
return 1;
}
auto decoded =
get_if<proto::ResultDecodedView>(&payload->response.result.value);
if (!decoded) {
if (!owned_handles.empty())
return 1;
auto error =
get_if<proto::ResultErrorView>(&payload->response.result.value);
fprintf(stderr, "dnimage: %.*s\n",
error ? int(error->error.message.size()) : 13,
error ? error->error.message.data() : "decode failed");
return 1;
}
const auto &pixmap = decoded->decoded.pixmap;
const uint64_t row = uint64_t(pixmap.width) * dawn::kBytesPerPixel;
if (!pixmap.width || !pixmap.height || pixmap.width > dawn::kMaxDimension ||
pixmap.height > dawn::kMaxDimension || pixmap.stride < row) {
fprintf(stderr, "dnimage: invalid pixmap geometry\n");
return 1;
}
const uint8_t *pixels = nullptr;
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();
}
auto cmm = make_shared<dawn::Cmm>();
auto icc = cmm->get_profile_sRGB(false)->to_bytes();
string error;
if (!write_png16(stdout, pixmap.width, pixmap.height, pixmap.stride, pixels,
icc, pixmap.orientation, &error)) {
fprintf(stderr, "dnimage: %s\n", error.c_str());
return 1;
}
}
+285
View File
@@ -0,0 +1,285 @@
//
// dnthumb.cpp: command-line client for dnthumbd
//
// Copyright The Dawn Authors
// SPDX-License-Identifier: MPL-2.0
//
#include "ipc-rpc.hpp"
#include "ipc-shm.hpp"
#include "ipc/thumbd.lxdr.hpp"
#include "libdn.h"
#include "png-io.hpp"
#include <charconv>
#include <chrono>
#include <cstdio>
#include <string_view>
#include <variant>
#include <vector>
#ifdef _WIN32
#include <fcntl.h>
#include <io.h>
#endif
using namespace std;
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)
{
const size_t x = value.find_first_of("xX");
if (x == string_view::npos)
return false;
auto a = from_chars(value.data(), value.data() + x, w);
auto b = from_chars(value.data() + x + 1, value.data() + value.size(), h);
return a.ec == errc{} && a.ptr == value.data() + x && b.ec == errc{} &&
b.ptr == value.data() + value.size() && w && h;
}
static bool
receive(dawn::ipc::Channel &channel, proto::FrameView &view,
vector<byte> &storage, vector<dawn::ipc::Handle> &handles,
chrono::milliseconds &budget)
{
if (!channel.recv(storage, budget, &handles))
return false;
dawn::ipc::Decoder decoder(storage);
return decode(decoder, view) && !decoder.remaining();
}
// --- Main --------------------------------------------------------------------
int
main(int argc, char **argv)
{
#ifdef _WIN32
_setmode(_fileno(stdin), _O_BINARY);
_setmode(_fileno(stdout), _O_BINARY);
#endif
// TODO(p): getopt_long
uint32_t bound_w = 512, bound_h = 256;
if (argc == 3 && string_view(argv[1]) == "-s") {
if (!parse_size(argv[2], bound_w, bound_h)) {
fprintf(stderr, "dnthumb: invalid size\n");
return 1;
}
} else if (argc != 1) {
fprintf(stderr, "Usage: dnthumb [-s WxH]\n");
return 1;
}
vector<uint8_t> input;
uint8_t chunk[65536];
while (size_t n = fread(chunk, 1, sizeof chunk, stdin)) {
if (uint64_t(input.size()) > uint64_t(proto::kThumbdMaxBlobSize) - n) {
fprintf(stderr, "dnthumb: input is too large\n");
return 1;
}
input.insert(input.end(), chunk, chunk + n);
}
if (ferror(stdin) || input.empty()) {
fprintf(stderr, "dnthumb: cannot read input\n");
return 1;
}
auto cmm = make_shared<dawn::Cmm>();
dawn::OpenContext context;
context.cmm = cmm;
context.screen_profile = cmm->get_profile_sRGB(false);
context.first_frame_only = true;
dawn::Error open_error;
auto image = dawn::open_from_data(input, context, &open_error);
if (!image) {
fprintf(stderr, "dnthumb: %s\n",
open_error.message.empty() ? "invalid PNG"
: open_error.message.c_str());
return 1;
}
int32_t orientation = int32_t(image->orientation);
if (orientation < 0 || orientation > 8)
orientation = 0;
if (auto it = image->text.find("dawn:orientation");
it != image->text.end()) {
int32_t parsed = 0;
auto result = from_chars(
it->second.data(), it->second.data() + it->second.size(), parsed);
if (result.ec == errc{} &&
result.ptr == it->second.data() + it->second.size() &&
parsed >= 0 && parsed <= 8)
orientation = parsed;
}
uint32_t display_w, display_h;
dawn::orientation_display_size(image->width, image->height,
dawn::orientation_or_0(dawn::Orientation(orientation)), &display_w,
&display_h);
const double scale =
min(1.0, min(double(bound_w) / display_w, double(bound_h) / display_h));
const uint32_t out_w = max(1u, uint32_t(display_w * scale + .5));
const uint32_t out_h = max(1u, uint32_t(display_h * scale + .5));
chrono::milliseconds budget = chrono::seconds(10);
auto channel = dawn::ipc::Channel::connect("thumbd", budget);
if (!channel.ok()) {
fprintf(stderr, "dnthumb: dnthumbd is not running\n");
return 1;
}
proto::Frame hello;
hello.payload.value =
proto::PayloadHello{proto::Hello{proto::kThumbdProtocolVersion, ""}};
if (!channel.send(encoded(hello), budget, {}))
return 1;
vector<byte> wire;
vector<dawn::ipc::Handle> reply_handles;
proto::FrameView reply;
const bool received = receive(channel, reply, wire, reply_handles, budget);
dawn::ipc::OwnedHandles owned_reply_handles(std::move(reply_handles));
if (!received || !owned_reply_handles.empty() ||
!holds_alternative<proto::PayloadHelloReplyView>(reply.payload.value)) {
fprintf(stderr, "dnthumb: incompatible daemon\n");
return 1;
}
const auto *accepted = get_if<proto::DaemonHelloReplyAcceptedView>(
&get<proto::PayloadHelloReplyView>(reply.payload.value)
.hello_reply.value);
if (!accepted || !accepted->limits.max_payload_size ||
accepted->limits.max_payload_size >
dawn::ipc::Connection::kMaxPayload ||
!accepted->limits.max_blob_size ||
accepted->limits.max_blob_size > proto::kThumbdMaxBlobSize) {
fprintf(stderr, "dnthumb: invalid daemon limits\n");
return 1;
}
const uint32_t max_payload = accepted->limits.max_payload_size;
const uint64_t max_blob = accepted->limits.max_blob_size;
channel.set_max_payload(max_payload);
proto::ScaleRequest scale_request;
scale_request.source.width = image->width;
scale_request.source.height = image->height;
scale_request.source.stride = image->stride;
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();
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))
return 1;
vector<dawn::ipc::Handle> handles;
proto::FrameView response;
const bool response_received =
receive(channel, response, wire, handles, budget);
dawn::ipc::OwnedHandles owned_handles(std::move(handles));
if (!response_received)
return 1;
auto payload = get_if<proto::PayloadResponseView>(&response.payload.value);
if (!payload)
return 1;
if (payload->response.id != 1) {
fprintf(stderr, "dnthumb: response request ID mismatch\n");
return 1;
}
auto scaled =
get_if<proto::ResultScaledView>(&payload->response.result.value);
if (!scaled) {
if (!owned_handles.empty())
return 1;
auto error =
get_if<proto::ResultErrorView>(&payload->response.result.value);
fprintf(stderr, "dnthumb: %.*s\n",
error ? int(error->error.message.size()) : 12,
error ? error->error.message.data() : "scale failed");
return 1;
}
const uint64_t rgba_size =
uint64_t(scaled->scaled.width) * scaled->scaled.height * 4;
if (!scaled->scaled.width || !scaled->scaled.height ||
scaled->scaled.width != out_w || scaled->scaled.height != out_h ||
scaled->scaled.width > dawn::kMaxDimension ||
scaled->scaled.height > dawn::kMaxDimension || rgba_size > max_blob) {
fprintf(stderr, "dnthumb: invalid scaled image geometry\n");
return 1;
}
span<const uint8_t> rgba;
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()};
}
vector<uint8_t> icc = image->effective_profile
? image->effective_profile->to_bytes()
: vector<uint8_t>{};
string error;
if (!write_png8(stdout, scaled->scaled.width, scaled->scaled.height, rgba,
icc, &error)) {
fprintf(stderr, "dnthumb: %s\n", error.c_str());
return 1;
}
}
+119
View File
@@ -0,0 +1,119 @@
//
// png-io.cpp: PNG output for daemon command-line clients
//
// Copyright The Dawn Authors
// SPDX-License-Identifier: MPL-2.0
//
#include "png-io.hpp"
#include <png.h>
#include <bit>
#include <cstdio>
#include <vector>
using namespace std;
static void
png_error_fn(png_structp png, png_const_charp message)
{
*(string *) png_get_error_ptr(png) = message;
longjmp(png_jmpbuf(png), 1);
}
static bool
begin(FILE *file, uint32_t width, uint32_t height, int depth,
span<const uint8_t> icc, string *error, png_structp *png_out,
png_infop *info_out)
{
png_structp png = png_create_write_struct(
PNG_LIBPNG_VER_STRING, error, png_error_fn, nullptr);
if (!png)
return false;
png_infop info = png_create_info_struct(png);
if (!info) {
png_destroy_write_struct(&png, nullptr);
return false;
}
if (setjmp(png_jmpbuf(png))) {
png_destroy_write_struct(&png, &info);
return false;
}
png_init_io(png, file);
png_set_IHDR(png, info, width, height, depth, PNG_COLOR_TYPE_RGBA,
PNG_INTERLACE_NONE, PNG_COMPRESSION_TYPE_BASE, PNG_FILTER_TYPE_BASE);
if (!icc.empty())
png_set_iCCP(png, info, "sRGB", PNG_COMPRESSION_TYPE_BASE,
const_cast<png_bytep>(icc.data()), png_uint_32(icc.size()));
*png_out = png;
*info_out = info;
return true;
}
bool
write_png16(FILE *file, uint32_t width, uint32_t height, uint32_t stride,
const uint8_t *pixels, span<const uint8_t> icc, int32_t orientation,
string *error)
{
png_structp png;
png_infop info;
if (!begin(file, width, height, 16, icc, error, &png, &info))
return false;
string text = to_string(orientation);
vector<uint16_t> row(size_t(width) * 4);
if (setjmp(png_jmpbuf(png))) {
png_destroy_write_struct(&png, &info);
return false;
}
png_text entry{};
entry.compression = PNG_TEXT_COMPRESSION_NONE;
entry.key = const_cast<char *>("dawn:orientation");
entry.text = text.data();
png_set_text(png, info, &entry, 1);
if constexpr (endian::native == endian::little)
png_set_swap(png);
png_write_info(png, info);
for (uint32_t y = 0; y < height; y++) {
const auto *src =
reinterpret_cast<const uint16_t *>(pixels + size_t(y) * stride);
for (uint32_t x = 0; x < width; x++) {
const uint32_t a = src[x * 4 + 3];
auto straight = [a](uint16_t v) {
return uint16_t(
a ? min(65535u, (uint32_t(v) * 65535u + a / 2) / a) : 0);
};
row[x * 4] = straight(src[x * 4 + 2]);
row[x * 4 + 1] = straight(src[x * 4 + 1]);
row[x * 4 + 2] = straight(src[x * 4]);
row[x * 4 + 3] = uint16_t(a);
}
png_write_row(png, reinterpret_cast<png_bytep>(row.data()));
}
png_write_end(png, info);
png_destroy_write_struct(&png, &info);
return true;
}
bool
write_png8(FILE *file, uint32_t width, uint32_t height,
span<const uint8_t> rgba, span<const uint8_t> icc, string *error)
{
if (rgba.size() < uint64_t(width) * height * 4)
return false;
png_structp png;
png_infop info;
if (!begin(file, width, height, 8, icc, error, &png, &info))
return false;
if (setjmp(png_jmpbuf(png))) {
png_destroy_write_struct(&png, &info);
return false;
}
png_write_info(png, info);
for (uint32_t y = 0; y < height; y++)
png_write_row(
png, const_cast<png_bytep>(rgba.data() + size_t(y) * width * 4));
png_write_end(png, info);
png_destroy_write_struct(&png, &info);
return true;
}
+20
View File
@@ -0,0 +1,20 @@
//
// png-io.hpp: PNG output for daemon command-line clients
//
// Copyright The Dawn Authors
// SPDX-License-Identifier: MPL-2.0
//
#pragma once
#include <cstdint>
#include <cstdio>
#include <span>
#include <string>
bool write_png16(FILE *file, uint32_t width, uint32_t height, uint32_t stride,
const uint8_t *bgra16_premul, std::span<const uint8_t> icc,
int32_t orientation, std::string *error);
bool write_png8(FILE *file, uint32_t width, uint32_t height,
std::span<const uint8_t> rgba, std::span<const uint8_t> icc,
std::string *error);