cannam@111: /* -*- c-basic-offset: 4 indent-tabs-mode: nil -*- vi:set ts=8 sts=4 sw=4: */ c@118: /* c@118: Piper C++ c@118: c@118: An API for audio analysis and feature extraction plugins. c@118: c@118: Centre for Digital Music, Queen Mary, University of London. c@118: Copyright 2006-2016 Chris Cannam and QMUL. c@118: c@118: Permission is hereby granted, free of charge, to any person c@118: obtaining a copy of this software and associated documentation c@118: files (the "Software"), to deal in the Software without c@118: restriction, including without limitation the rights to use, copy, c@118: modify, merge, publish, distribute, sublicense, and/or sell copies c@118: of the Software, and to permit persons to whom the Software is c@118: furnished to do so, subject to the following conditions: c@118: c@118: The above copyright notice and this permission notice shall be c@118: included in all copies or substantial portions of the Software. c@118: c@118: THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, c@118: EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF c@118: MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND c@118: NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS BE LIABLE FOR c@118: ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF c@118: CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION c@118: WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE. c@118: c@118: Except as contained in this notice, the names of the Centre for c@118: Digital Music; Queen Mary, University of London; and Chris Cannam c@118: shall not be used in advertising or otherwise to promote the sale, c@118: use or other dealings in this Software without prior written c@118: authorization. c@118: */ c@96: c@96: #ifndef PIPER_CAPNP_CLIENT_H c@96: #define PIPER_CAPNP_CLIENT_H c@96: c@96: #include "Loader.h" c@96: #include "PluginClient.h" c@96: #include "PluginStub.h" c@96: #include "SynchronousTransport.h" c@96: c@96: #include "vamp-support/AssignedPluginHandleMapper.h" c@96: #include "vamp-capnp/VampnProto.h" c@96: c@96: #include c@96: c@97: namespace piper_vamp { c@97: namespace client { c@96: c@100: /** c@100: * Client for a request-response Piper server, i.e. using the c@100: * RpcRequest/RpcResponse structures with a single process call rather c@100: * than having individual RPC methods, with a synchronous transport c@100: * such as a subprocess pipe arrangement. Only one request can be c@100: * handled at a time. This class is thread-safe if and only if it is c@100: * constructed with a thread-safe SynchronousTransport implementation. c@100: */ c@96: class CapnpRRClient : public PluginClient, c@118: public Loader c@96: { c@96: // unsigned to avoid undefined behaviour on possible wrap c@96: typedef uint32_t ReqId; c@96: c@96: class CompletenessChecker : public MessageCompletenessChecker { c@96: public: c@96: bool isComplete(const std::vector &message) const override { c@96: auto karr = toKJArray(message); c@96: size_t words = karr.size(); c@96: size_t expected = capnp::expectedSizeInWordsFromPrefix(karr); c@96: if (words > expected) { c@96: std::cerr << "WARNING: obtained more data than expected (" c@96: << words << " " << sizeof(capnp::word) c@96: << "-byte words, expected " c@96: << expected << ")" << std::endl; c@96: } c@96: return words >= expected; c@96: } c@96: }; c@96: c@96: public: c@96: CapnpRRClient(SynchronousTransport *transport) : //!!! ownership? shared ptr? c@96: m_transport(transport), c@96: m_completenessChecker(new CompletenessChecker) { c@96: transport->setCompletenessChecker(m_completenessChecker); c@96: } c@96: c@96: ~CapnpRRClient() { c@96: delete m_completenessChecker; c@96: } c@96: c@96: //!!! obviously, factor out all repetitive guff c@96: c@96: //!!! list and load are supposed to be called by application code, c@96: //!!! but the rest are only supposed to be called by the plugin -- c@96: //!!! sort out the api here c@96: c@96: // Loader methods: c@96: c@97: ListResponse c@96: listPluginData() override { c@96: c@96: if (!m_transport->isOK()) { c@96: throw std::runtime_error("Piper server failed to start"); c@96: } c@96: c@96: capnp::MallocMessageBuilder message; c@118: piper::RpcRequest::Builder builder = message.initRoot(); c@96: VampnProto::buildRpcRequest_List(builder); c@96: ReqId id = getId(); c@96: builder.getId().setNumber(id); c@96: c@118: auto karr = call(message); c@96: c@96: capnp::FlatArrayMessageReader responseMessage(karr); c@97: piper::RpcResponse::Reader reader = responseMessage.getRoot(); c@96: c@97: checkResponseType(reader, piper::RpcResponse::Response::Which::LIST, id); c@96: c@97: ListResponse lr; c@96: VampnProto::readListResponse(lr, reader.getResponse().getList()); c@96: return lr; c@96: } c@96: c@97: LoadResponse c@97: loadPlugin(const LoadRequest &req) override { c@96: c@96: if (!m_transport->isOK()) { c@96: throw std::runtime_error("Piper server failed to start"); c@96: } c@96: c@97: LoadResponse resp; c@96: PluginHandleMapper::Handle handle = serverLoad(req.pluginKey, c@96: req.inputSampleRate, c@96: req.adapterFlags, c@96: resp.staticData, c@96: resp.defaultConfiguration); c@96: c@96: Vamp::Plugin *plugin = new PluginStub(this, c@96: req.pluginKey, c@96: req.inputSampleRate, c@96: req.adapterFlags, c@96: resp.staticData, c@96: resp.defaultConfiguration); c@96: c@96: m_mapper.addPlugin(handle, plugin); c@96: c@96: resp.plugin = plugin; c@96: return resp; c@96: } c@96: c@96: // PluginClient methods: c@96: c@96: virtual c@96: Vamp::Plugin::OutputList c@96: configure(PluginStub *plugin, c@97: PluginConfiguration config) override { c@96: c@96: if (!m_transport->isOK()) { c@96: throw std::runtime_error("Piper server failed to start"); c@96: } c@96: c@97: ConfigurationRequest request; c@96: request.plugin = plugin; c@96: request.configuration = config; c@96: c@96: capnp::MallocMessageBuilder message; c@97: piper::RpcRequest::Builder builder = message.initRoot(); c@96: c@96: VampnProto::buildRpcRequest_Configure(builder, request, m_mapper); c@96: ReqId id = getId(); c@96: builder.getId().setNumber(id); c@96: c@118: auto karr = call(message); c@96: c@96: capnp::FlatArrayMessageReader responseMessage(karr); c@97: piper::RpcResponse::Reader reader = responseMessage.getRoot(); c@96: c@96: //!!! handle (explicit) error case c@96: c@97: checkResponseType(reader, piper::RpcResponse::Response::Which::CONFIGURE, id); c@96: c@97: ConfigurationResponse cr; c@96: VampnProto::readConfigurationResponse(cr, c@96: reader.getResponse().getConfigure(), c@96: m_mapper); c@96: c@96: return cr.outputs; c@96: }; c@96: c@96: virtual c@96: Vamp::Plugin::FeatureSet c@96: process(PluginStub *plugin, c@96: std::vector > inputBuffers, c@96: Vamp::RealTime timestamp) override { c@96: c@96: if (!m_transport->isOK()) { c@96: throw std::runtime_error("Piper server failed to start"); c@96: } c@96: c@97: ProcessRequest request; c@96: request.plugin = plugin; c@96: request.inputBuffers = inputBuffers; c@96: request.timestamp = timestamp; c@96: c@96: capnp::MallocMessageBuilder message; c@97: piper::RpcRequest::Builder builder = message.initRoot(); c@96: VampnProto::buildRpcRequest_Process(builder, request, m_mapper); c@118: ReqId id = getId(); c@96: builder.getId().setNumber(id); c@96: c@118: auto karr = call(message); c@96: c@96: capnp::FlatArrayMessageReader responseMessage(karr); c@97: piper::RpcResponse::Reader reader = responseMessage.getRoot(); c@96: c@96: //!!! handle (explicit) error case c@96: c@97: checkResponseType(reader, piper::RpcResponse::Response::Which::PROCESS, id); c@96: c@97: ProcessResponse pr; c@96: VampnProto::readProcessResponse(pr, c@96: reader.getResponse().getProcess(), c@96: m_mapper); c@96: c@96: return pr.features; c@96: } c@96: c@96: virtual Vamp::Plugin::FeatureSet c@96: finish(PluginStub *plugin) override { c@96: c@96: if (!m_transport->isOK()) { c@96: throw std::runtime_error("Piper server failed to start"); c@96: } c@96: c@97: FinishRequest request; c@96: request.plugin = plugin; c@96: c@96: capnp::MallocMessageBuilder message; c@97: piper::RpcRequest::Builder builder = message.initRoot(); c@96: c@96: VampnProto::buildRpcRequest_Finish(builder, request, m_mapper); c@96: ReqId id = getId(); c@96: builder.getId().setNumber(id); c@96: c@118: auto karr = call(message); c@96: c@96: capnp::FlatArrayMessageReader responseMessage(karr); c@97: piper::RpcResponse::Reader reader = responseMessage.getRoot(); c@96: c@96: //!!! handle (explicit) error case c@96: c@97: checkResponseType(reader, piper::RpcResponse::Response::Which::FINISH, id); c@96: c@97: FinishResponse pr; c@96: VampnProto::readFinishResponse(pr, c@96: reader.getResponse().getFinish(), c@96: m_mapper); c@96: c@96: m_mapper.removePlugin(m_mapper.pluginToHandle(plugin)); c@96: c@118: // Don't delete the plugin. It's the plugin that is supposed c@118: // to be calling us here c@96: c@96: return pr.features; c@96: } c@96: c@96: virtual void c@96: reset(PluginStub *plugin, c@97: PluginConfiguration config) override { c@96: c@96: // Reload the plugin on the server side, and configure it as requested c@96: c@96: if (!m_transport->isOK()) { c@96: throw std::runtime_error("Piper server failed to start"); c@96: } c@96: c@96: if (m_mapper.havePlugin(plugin)) { c@96: (void)finish(plugin); // server-side unload c@96: } c@96: c@97: PluginStaticData psd; c@97: PluginConfiguration defaultConfig; c@96: PluginHandleMapper::Handle handle = c@96: serverLoad(plugin->getPluginKey(), c@96: plugin->getInputSampleRate(), c@96: plugin->getAdapterFlags(), c@96: psd, defaultConfig); c@96: c@96: m_mapper.addPlugin(handle, plugin); c@96: c@96: (void)configure(plugin, config); c@96: } c@96: c@96: private: c@96: AssignedPluginHandleMapper m_mapper; c@96: ReqId getId() { c@96: //!!! todo: mutex c@96: static ReqId m_nextId = 0; c@96: return m_nextId++; c@96: } c@96: c@96: static c@96: kj::Array c@96: toKJArray(const std::vector &buffer) { c@118: // We could do this whole thing with fewer copies, but let's c@118: // see whether it matters first c@96: size_t wordSize = sizeof(capnp::word); c@118: size_t words = buffer.size() / wordSize; c@118: kj::Array karr(kj::heapArray(words)); c@118: memcpy(karr.begin(), buffer.data(), words * wordSize); c@118: return karr; c@96: } c@96: c@96: void c@97: checkResponseType(const piper::RpcResponse::Reader &r, c@97: piper::RpcResponse::Response::Which type, c@96: ReqId id) { c@96: c@96: if (r.getResponse().which() != type) { c@118: std::cerr << "checkResponseType: wrong response type (received " c@118: << int(r.getResponse().which()) << ", expected " c@118: << int(type) << ")" c@118: << std::endl; c@96: throw std::runtime_error("Wrong response type"); c@96: } c@96: if (ReqId(r.getId().getNumber()) != id) { c@118: std::cerr << "checkResponseType: wrong response id (received " c@118: << r.getId().getNumber() << ", expected " << id << ")" c@118: << std::endl; c@96: throw std::runtime_error("Wrong response id"); c@96: } c@96: } c@96: c@96: kj::Array c@96: call(capnp::MallocMessageBuilder &message) { c@96: auto arr = capnp::messageToFlatArray(message); c@96: auto responseBuffer = m_transport->call(arr.asChars().begin(), c@96: arr.asChars().size()); c@118: return toKJArray(responseBuffer); c@96: } c@96: c@96: PluginHandleMapper::Handle c@96: serverLoad(std::string key, float inputSampleRate, int adapterFlags, c@97: PluginStaticData &psd, c@97: PluginConfiguration &defaultConfig) { c@96: c@97: LoadRequest request; c@96: request.pluginKey = key; c@96: request.inputSampleRate = inputSampleRate; c@96: request.adapterFlags = adapterFlags; c@96: c@96: capnp::MallocMessageBuilder message; c@97: piper::RpcRequest::Builder builder = message.initRoot(); c@96: c@96: VampnProto::buildRpcRequest_Load(builder, request); c@96: ReqId id = getId(); c@96: builder.getId().setNumber(id); c@96: c@118: auto karr = call(message); c@96: c@96: //!!! ... --> will also need some way to kill this process c@96: //!!! (from another thread) c@96: c@96: capnp::FlatArrayMessageReader responseMessage(karr); c@97: piper::RpcResponse::Reader reader = responseMessage.getRoot(); c@96: c@96: //!!! handle (explicit) error case c@96: c@97: checkResponseType(reader, piper::RpcResponse::Response::Which::LOAD, id); c@96: c@97: const piper::LoadResponse::Reader &lr = reader.getResponse().getLoad(); c@96: VampnProto::readExtractorStaticData(psd, lr.getStaticData()); c@96: VampnProto::readConfiguration(defaultConfig, lr.getDefaultConfiguration()); c@96: return lr.getHandle(); c@96: }; c@96: c@96: private: c@96: SynchronousTransport *m_transport; //!!! I don't own this, but should I? c@96: CompletenessChecker *m_completenessChecker; // I own this c@96: }; c@96: c@96: } c@96: } c@96: c@96: #endif