looper/playback_process.cpp

#include <google/protobuf/descriptor.h>
#include <google/protobuf/wire_format_lite.h>
#include <grpcpp/client_context.h>
#include <grpcpp/impl/codegen/server_interface.h>
#include <grpcpp/server_builder.h>
#include <grpcpp/support/status.h>
#include "playback_process.hpp"
#include "ipc/common.pb.h"
#include "playback_backend.hpp"
#include <functional>
#include "rpc.hpp"
#include "thirdparty/CRC.hpp"
#include <stdlib.h>
#include <string.h>
#include <stdio.h>
#include <unistd.h>
#include <exception>
#include <thread>
#include <config.h>
#include <signal.h>
#include "backend.hpp"
#include "util.hpp"
#include "log.hpp"
#include <chrono>
#include <fmt/core.h>
#include <fmt/format.h>
#ifdef __WINDOWS__
#include <windows.h>
#endif
#include <google/protobuf/message.h>
#include "util.hpp"
using namespace google::protobuf;
int sndfd;
int rcvfd;
extern char *executable_path;
using grpc::ClientContext;
void print_field_descriptor(const google::protobuf::FieldDescriptor *fdesc, const google::protobuf::Message &msg, size_t level) {
    auto desc = msg.GetDescriptor();
    auto reflection = msg.GetReflection();
    static std::map<google::protobuf::FieldDescriptor::Label, const char*> labelNames = {
        {(google::protobuf::FieldDescriptor::Label)0, "(Error)"},
        {google::protobuf::FieldDescriptor::LABEL_OPTIONAL, "Optional"},
        {google::protobuf::FieldDescriptor::LABEL_REPEATED, "Repeated"},
        {google::protobuf::FieldDescriptor::LABEL_REQUIRED, "Required"}
    };
    DEBUG.writef_level(level, "- %s = %d", fdesc->full_name().c_str(), fdesc->number());
    level++;
    DEBUG.writef_level(level, "Type: %s (C++: %s)", fdesc->type_name(), fdesc->cpp_type_name());
    std::string value = "(None)";
    if (fdesc->has_default_value()) {
        FORMAT_CPP_TYPE_LOWERCASE(fdesc, value, fdesc->default_value_);
        DEBUG.writef_level(level, "Default value: %s", value.c_str());
    }
    DEBUG.writef_level(level, "Label: %s", labelNames.contains(fdesc->label()) ? labelNames[fdesc->label()] : labelNames[(google::protobuf::FieldDescriptor::Label)0]);
    DEBUG.writef_level(level, "Index: %d", fdesc->index());
    if (fdesc->type() == google::protobuf::FieldDescriptor::TYPE_MESSAGE || fdesc->type() == google::protobuf::FieldDescriptor::TYPE_GROUP) {
        const google::protobuf::Descriptor *desc = fdesc->message_type();
        DEBUG.writef_level(level, "Value: (Message)");
        const Message &out_msg = reflection->GetMessage(msg, fdesc);
        print_ipc_message(out_msg, level + 1);
        return;
    }
    if (fdesc->type() == google::protobuf::FieldDescriptor::TYPE_BYTES) {
        value = "(bytes)";
    } else {
        FORMAT_CPP_TYPE_TITLECASE(fdesc, value, reflection->Get, msg, fdesc);
    }
    DEBUG.writef_level(level, "Value: %s", value.c_str());
}
void print_ipc_message(const google::protobuf::Message &msg, size_t level) {
    const google::protobuf::Descriptor *msgdesc = msg.GetDescriptor();
    const google::protobuf::Reflection *reflect = msg.GetReflection();
    DEBUG.writef_level(level, "Message type: %s", msg.GetTypeName().c_str());
    std::vector<const FieldDescriptor *> descriptors;
    reflect->ListFields(msg, &descriptors);
    DEBUG.writef_level(level, "Message field count: %d", descriptors.size());
    for (auto field : descriptors) {
        print_field_descriptor(field, msg, level);
    }
    auto &unknown_fields = reflect->GetUnknownFields(msg);
    DEBUG.writef_level(level, "Unknown fields: %d", unknown_fields.field_count());
}
void show_command(const std::string &cmdid, const google::protobuf::Message &msg) {
	DEBUG.writefln("Command %s:", cmdid.c_str());
	print_ipc_message(msg);
}
grpc::Status PlaybackProcessServiceImpl::Render(grpc::ServerContext *ctx, const RenderCommand *cmd, RenderResponseOrError *response) {
    size_t maxlen = cmd->len();
    void *ptr = malloc(maxlen);
    size_t len = cur_backend->render(ptr, maxlen);
    if (len == 0) {
    	DEBUG.writeln("Didn't get any audio when rendering");
    } else if (is_zeroes(ptr, len)) {
    	DEBUG.writeln("Buffer was only zeroes!");
    }
    auto output = new RenderResponse();
    output->set_data((const char*)ptr, len);
    free(ptr);
    output->set_len(len);
    response->set_allocated_output(output);
    return grpc::Status::OK;
}
grpc::Status PlaybackProcessServiceImpl::Get(grpc::ServerContext *ctx, const GetProperty *request, PropertyDataOrError *response) {
    PropertyData *data = new PropertyData();
    ErrorResponse *err_maybe = nullptr;
    switch (request->id()) {
        case PropertyId::StreamIdProperty: {
            StreamId id;
            id.set_id(cur_backend->get_stream_idx());
            data->mutable_value()->PackFrom(id);
        } break;
        case PropertyId::FilenameProperty: {
            StringProperty prop;
            prop.set_value(std::filesystem::path(cur_backend->get_current_file().value_or("None")).filename().string());
            data->mutable_value()->PackFrom(prop);
        } break;
        case PropertyId::BackendId: {
            StringProperty prop;
            prop.set_value(cur_backend->get_id());
            data->mutable_value()->PackFrom(prop);
        } break;
        case PropertyId::BackendName: {
            StringProperty prop;
            prop.set_value(cur_backend->get_name());
            data->mutable_value()->PackFrom(prop);
        }
        case PropertyId::FilePathProperty: {
            StringProperty path;
            path.set_value(cur_backend->get_current_file().value_or(""));
            data->mutable_value()->PackFrom(path);
        } break;
        case PropertyId::SpecProperty: {
            AudioSpec spec;
            SDL_AudioSpec sdl_spec = cur_backend->get_spec();
            audio_data_t sample_spec = sdl_to_sample_spec(sdl_spec.format);
            DEBUG.writefln("Sample_spec.size: %d", sample_spec.size);
            spec.set_bits(sample_spec.size * 8);
            spec.set_channel_count(sdl_spec.channels);
            spec.set_endian(sample_spec.endian ? EndianID::BIG : EndianID::LITTLE);
            spec.set_format_type(sample_spec.is_float ? FormatType::FLOAT : sample_spec.is_signed ? FormatType::SIGNED : FormatType::UNSIGNED);
            spec.set_sample_rate(sdl_spec.freq);
            data->mutable_value()->PackFrom(spec);
        } break;
        case PropertyId::StreamsProperty: {
            auto full_streams = cur_backend->get_streams();
            if (request->has_idx()) {
                size_t idx = (size_t)request->idx();
                Stream stream;
                if (idx >= full_streams.size()) {
                    ErrorResponse *err = new ErrorResponse();
                    err->set_id("invalid_index");
                    err->set_desc("An attempt to access an array element outside the boundaries of that array was detected.");
                    err->set_fatal(false);
                    err_maybe = err;
                    break;
                }
                auto backend_stream = full_streams[idx];
                stream.set_id(backend_stream.id);
                stream.set_len(backend_stream.length);
                stream.set_title(backend_stream.name);
                data->mutable_value()->PackFrom(stream);
            } else {
                StreamList out_streams;
                for (auto backend_stream : full_streams) {
                    Stream stream = *out_streams.add_streams();
                    stream.set_id(backend_stream.id);
                    stream.set_len(backend_stream.length);
                    stream.set_title(backend_stream.name);
                }
                data->mutable_value()->PackFrom(out_streams);
            }
        } break;
        case PropertyId::TitleProperty: {
            StringProperty title;
            title.set_value(cur_backend->get_title().value_or(""));
            data->mutable_value()->PackFrom(title);
        } break;
        case PropertyId::PositionProperty: {
            Position pos;
            pos.set_pos(cur_backend->get_position());
            data->mutable_value()->PackFrom(pos);
        } break;
        case PropertyId::BackendSpecific: {
            if (!request->has_path()) {
                ErrorResponse *err = new ErrorResponse();
                err->set_id("param_missing");
                err->set_fatal(false);
                err->set_desc("Backend specific parameters require a path.");
                err_maybe = err;
                break;
            }
            auto output_maybe = cur_backend->get(request->path());
            if (output_maybe.has_value()) {
                data->mutable_value()->PackFrom(output_maybe.value());
            } else {
                ErrorResponse *err = new ErrorResponse;
                err->set_id("not_found");
                err->set_fatal(false);
                err->set_desc("The backend reported that the property was not found.");
                err_maybe = err;
            }
        } break;
        default: {
            ErrorResponse *err = new ErrorResponse();
            err->set_id("invalid_property");
            err->set_fatal(false);
            err->set_desc("The property requested was invalid or write-only");
            err_maybe = err;
        } break;
    }
    if (err_maybe != nullptr) {
        delete data;
        response->set_allocated_err(err_maybe);
    } else {
        response->set_allocated_output(data);
    }
    return grpc::Status::OK;
}
grpc::Status HostProcessImpl::WriteLog(grpc::ServerContext *ctx, const LogMessage *msg, SimpleAckResponse *response) {
    return grpc::Status::OK;
}
grpc::Status HostProcessImpl::SetAddress(grpc::ServerContext *ctx, const StringProperty *value, MaybeError *response) {
    process->host_channel.value().construct_client(value->value());
    process->started.notify_all();
    return grpc::Status::OK;
}
grpc::Status PlaybackProcessServiceImpl::Set(grpc::ServerContext *ctx, const SetProperty *request, MaybeError *response) {
    ErrorResponse *err_maybe = nullptr;
    switch (request->id()) {
        case PropertyId::PositionProperty: {
            cur_backend->seek(resolve_any<Position>(request->value())->pos());
        } break;
        case PropertyId::StreamIdProperty: {
            cur_backend->switch_stream(resolve_any<StreamId>(request->value())->id());
        } break;
        case PropertyId::BackendSpecific: {
            if (!request->has_path()) {
                ErrorResponse *err = new ErrorResponse();
                err->set_id("param_missing");
                err->set_fatal(false);
                err->set_desc("Backend specific parameters require a path.");
                err_maybe = err;
                break;
            }
            if (!cur_backend->set(request->path(), request->value())) {
                ErrorResponse *err = new ErrorResponse();
                err->set_id("not_found");
                err->set_fatal(false);
                err->set_desc("The backend reported that the property being set was invalid for it.");
                err_maybe = err;
            }
        } break;
        default: {
            ErrorResponse *err = new ErrorResponse();
            err->set_id("invalid_property");
            err->set_fatal(false);
            err->set_desc("The property requested was invalid or read-only");
            err_maybe = err;
        } break;
    }
    if (err_maybe != nullptr) {
        response->set_allocated_err(err_maybe);
    }
    return grpc::Status::OK;
}
grpc::Status PlaybackProcessServiceImpl::Reset(grpc::ServerContext *ctx, const ResetProperty *request, ResetResponse *response) {
    auto path = request->path();
    std::optional<uint64_t> idx = {};
    if (request->has_idx()) {
        idx = request->idx();
    }
    auto output = cur_backend->reset(path);
    if (output.has_value()) {
        PropertyDataOrError *real_output = new PropertyDataOrError();
        PropertyData *data = new PropertyData();
        data->mutable_value()->PackFrom(output.value());
        data->set_id(PropertyId::BackendSpecific);
        data->clear_idx();
        real_output->set_allocated_output(data);
        response->set_allocated_value(real_output);
    }
    return grpc::Status::OK;
}
grpc::Status PlaybackProcessServiceImpl::Quit(grpc::ServerContext *ctx, const QuitCmd *request, MaybeError *response) {
    process->done = true;
    process->playback_process_channel.value().get_server()->get_server()->Shutdown();
    return grpc::Status::OK;
}
grpc::Status PlaybackProcessServiceImpl::Init(grpc::ServerContext *ctx, const InitCommand *cmd, MaybeError *response) {
    cur_backend = nullptr;
    auto filename = cmd->filename();
    auto idx = cmd->idx();
    for (auto &backend : PlaybackBackendHelper()) {
        DEBUG.writefln("Trying backend: %s", backend.second->get_name().c_str());
        try {
            backend.second->init(filename.c_str(), idx);
        } catch (std::exception e) {
            DEBUG.writeln("Cleaning up backend.");
            backend.second->cleanup();
            continue;
        }
        cur_backend = backend.second;
        DEBUG.writefln("Using backend: %s", backend.second->get_name().c_str());
        break;
    }
    if (cur_backend == nullptr) {
        ErrorResponse *maybe_error = new ErrorResponse();
        maybe_error->set_desc("Couldn't find a backend.");
        maybe_error->set_id("no_backend_for_file");
        maybe_error->set_fatal(true);
        response->set_allocated_err(maybe_error);
        process->done = true;
        process->playback_process_channel.value().get_server()->get_server()->Shutdown();
        DEBUG.writefln("Couldn't find any backend.");
        return grpc::Status::OK;
    }
    return grpc::Status::OK;
}

void init_logging_subprocess(PlaybackProcess *proc) {
    Looper::Log::init_logging();
}
PlaybackStream deserialize_stream(Stream serialized) {
    PlaybackStream stream;
    stream.id = serialized.id();
    stream.length = serialized.len();
    stream.name = serialized.title();
    return stream;
}
Stream serialize_stream(PlaybackStream stream, std::optional<Stream*> stream_ptr = {}) {
    Stream serialized;
    serialized.set_id(stream.id);
    serialized.set_len(stream.length);
    serialized.set_title(stream.name);
    if (stream_ptr.has_value()) {
        *(stream_ptr.value()) = serialized;
    }
    return serialized;
}
std::vector<PlaybackStream> deserialize_stream_list(StreamList list) {
    int len = list.streams_size();
    std::vector<PlaybackStream> output;
    output.reserve(len);
    for (int i = 0; i < len; i++) {
        Stream stream = list.streams(i);
        output.push_back(deserialize_stream(stream));
    }
    return output;
}
StreamList serialize_stream_list(std::vector<PlaybackStream> streams) {
    StreamList list;
    for (auto input : streams) {
        Stream *stream = list.add_streams();
        serialize_stream(input, stream);
    }
    return list;
}
PlaybackProcess::PlaybackProcess(std::vector<std::string> args) {
    done = false;
    is_playback_process = true;
    Looper::Log::init_logging();
    init_playback_backends();
    DEBUG.writeln("Playback backends: ");
    for (auto &backend : PlaybackBackendHelper()) {
    	DEBUG.writefln(" - %s", backend.second->get_id().c_str());
    }
    init_audio_data();
    std::string address = args[0];
    auto mk_service = [this]() -> grpc::Service* {
        PlaybackProcessServiceImpl *output = new PlaybackProcessServiceImpl();
        output->process = this;
        return output;
    };
    playback_process_channel = IPCChannel<PlaybackProcessService, HostProcess>(mk_service);
    playback_process_channel.value().construct_client(address);
    DEBUG.writefln("Host process address: %s", address.c_str());
    {
        StringProperty property;
        property.set_value(playback_process_channel.value().get_server()->get_address());
        ClientContext ctx;
        MaybeError response;
        playback_process_channel.value().get_stub()->SetAddress(&ctx, property, &response);
    }
}
PlaybackProcess::PlaybackProcess(std::string filename, int idx) {
    done = false;

    auto mk_service = [this]() -> grpc::Service* {
        HostProcessImpl *output = new HostProcessImpl();
        output->process = this;
        return output;
    };
    this->done = false;
    host_channel = IPCChannel<HostProcess, PlaybackProcessService>(mk_service);
    std::string address = host_channel.value().get_server()->get_address();
    std::vector<std::string> new_args;
    new_args.push_back(executable_path);
    new_args.push_back("--process-type");
    new_args.push_back("playback");
    new_args.push_back(address);
    pid = launch(new_args);
    std::thread process_check_thread(std::mem_fn(&PlaybackProcess::threadfunc), this);
    std::unique_lock lk(start_mutex);
    started.wait(lk);
    if (done) {
        ERROR.writeln("Playback process exited too early!");
        throw std::exception();
    }
    lk.unlock();
    done = true;
    process_check_thread.join();
    DEBUG.writeln("Playback process started.");
    ClientContext ctx;
    InitCommand cmd;
    cmd.set_filename(filename);
    cmd.set_idx(idx);
    MaybeError output;
    get_stub()->Init(&ctx, cmd, &output);
    if (output.has_err()) {
        throw std::exception();
    }
}
bool PlaybackProcess::process_running() {
    if (is_playback_process) return true;
    return kill(pid, 0) == 0;
}
void PlaybackProcess::run_playback_process() {
    playback_process_channel.value().get_server()->get_server()->Wait();
}
int looper_run_playback_process(std::vector<std::string> args) {
    auto proc = PlaybackProcess(args);
    proc.run_playback_process();
    return 0;
}
PropertyData PlaybackProcess::get_property(PropertyId property, std::optional<uint64_t> idx) {
    GetProperty get_property;
    get_property.set_id(property);
    if (idx.has_value()) {
        get_property.set_idx(idx.value());
    } else {
        get_property.clear_idx();
    }
    ClientContext context;
    PropertyDataOrError output;
    get_stub()->Get(&context, get_property, &output);
    if (output.has_err()) {
        throw std::exception();
    }
    return output.output();
}
void PlaybackProcess::set_property(PropertyId id, PropertyData data, std::optional<uint64_t> idx) {
    SetProperty set_property;
    set_property.set_id(id);
    set_property.mutable_value()->PackFrom(data);
    if (idx.has_value()) {
        set_property.set_idx(idx.value());
    } else {
        set_property.clear_idx();
    }
    ClientContext context;
    MaybeError output;
    get_stub()->Set(&context, set_property, &output);
}
std::string PlaybackProcess::get_version_code() {
    return std::string(TAG) +
        #ifdef DEBUG_MODE
        std::string("-debugmode") +
        #endif
        std::string("-at") + std::string(__TIME__);
}
double PlaybackProcess::get_position() {
    auto *tmp = get_property_value<Position>(PropertyId::PositionProperty);
    double output = tmp->pos();
    delete tmp;
    return output;
}
void PlaybackProcess::set_position(double value) {
    Position *pos = new Position();
    pos->set_pos(value);
    set_property_value<Position>(PropertyId::PositionProperty, pos);
    delete pos;
}
size_t PlaybackProcess::get_stream_idx() {
    StreamId *id = get_property_value<StreamId>(PropertyId::StreamIdProperty);
    size_t output = id->id();
    delete id;
    return output;
}
void PlaybackProcess::set_stream_idx(size_t idx) {
    StreamId *id = new StreamId();
    id->set_id(idx);
    set_property_value<StreamId>(PropertyId::StreamIdProperty, id);
    delete id;
}
std::string PlaybackProcess::get_title() {
    return get_property_string(PropertyId::TitleProperty);
}
std::string PlaybackProcess::get_file_path() {
    return get_property_string(PropertyId::FilePathProperty);
}
std::string PlaybackProcess::get_file_name() {
    return get_property_string(PropertyId::FilenameProperty);
}
PlaybackStream PlaybackProcess::get_playback_stream(size_t idx) {
    auto *stream = get_property_value<Stream>(PropertyId::StreamsProperty, idx);
    PlaybackStream output = deserialize_stream(*stream);
    delete stream;
    return output;
}
std::string PlaybackProcess::get_backend_id() {
    return get_property_string(PropertyId::BackendId);
}
std::string PlaybackProcess::get_backend_name() {
    return get_property_string(PropertyId::BackendName);
}
std::vector<PlaybackStream> PlaybackProcess::get_playback_streams() {
    auto *list = get_property_value<StreamList>(PropertyId::StreamsProperty);
    auto output = deserialize_stream_list(*list);
    delete list;
    return output;
}
size_t PlaybackProcess::render(void *buf, size_t maxlen) {
    RenderCommand rend_cmd = RenderCommand();
    rend_cmd.set_len(maxlen);
    ClientContext ctx;
    RenderResponseOrError response;
    get_stub()->Render(&ctx, rend_cmd, &response);
    if (response.has_err()) {
    	ERROR.writefln("Error rendering audio: %s", response.err().id().c_str());
        return 0;
    } else {
    	std::string data = response.output().data();
    	if (data.length() == 0) {
	    	WARNING.writeln("Rendering audio didn't produce anything!");
		} else if (is_zeroes(data.data(), data.length())) {
	    	DEBUG.writeln("RECV'd buffer was only zeroes!");
	    }
        memcpy(buf, data.data(), data.length());
        return data.length();
    }
}
AudioSpec *PlaybackProcess::get_audio_spec() {
    return get_property_value<AudioSpec>(PropertyId::SpecProperty);
}
PlaybackProcess::~PlaybackProcess() {
    ClientContext ctx;
    QuitCmd quit_cmd;
    MaybeError output;
    get_stub()->Quit(&ctx, quit_cmd, &output);
    done = true;
}
void PlaybackProcess::threadfunc() {
    while (!process_running()) {
        std::this_thread::yield();
        if (done) return;
    }
    while (true) {
        if (!process_running()) {
            done = true;
            started.notify_all();
        }
        if (done) {
            return;
        }
        std::this_thread::yield();
    }
}