vulkancts/vkscserver/vksIPC.cpp

/*-------------------------------------------------------------------------
 * Vulkan CTS Framework
 * --------------------
 *
 * Copyright (c) 2021 The Khronos Group Inc.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *      http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 *
 *-------------------------------------------------------------------------*/

#include "vksIPC.hpp"
#include "vksStore.hpp"
#include "vksClient.hpp"

#include <future>

namespace vksc_server
{

namespace ipc
{

constexpr int DefaultPortIPC = 57323;

struct ChildConnection
{
    int id;
    std::unique_ptr<de::Socket> socket;
    std::atomic<bool> &appactive;
    vector<u8> recvb;
};

struct PacketsLoop
{
    ChildConnection client;
    Store &fileStore;

    template <typename T>
    static void SendResponse(ChildConnection &c, T &data)
    {
        SendPayloadWithHeader(c.socket.get(), T::Type(), Serialize(data));
    }

    void ProcessPacketsOnServer(ChildConnection &c, u32 type, vector<u8> packet)
    {
        switch (type)
        {
        case StoreContentRequest::Type():
        {
            auto req = Deserialize<StoreContentRequest>(packet);
            bool ok  = fileStore.Set(req.name, req.data);

            StoreContentResponse res;
            res.status = ok;
            SendResponse(c, res);
        }
        break;
        case GetContentRequest::Type():
        {
            auto req = Deserialize<GetContentRequest>(packet);

            vector<u8> content;
            bool ok = fileStore.Get(req.path, content, req.removeAfter);

            GetContentResponse res;
            res.status = ok;
            res.data   = std::move(content);
            SendResponse(c, res);
        }
        break;

        default:
            throw std::runtime_error("ipc communication error");
        }
    }

    void Loop()
    {
        while (client.socket->isConnected() && client.appactive)
        {
            RecvSome(client.socket.get(), client.recvb);
            auto interpret = [this](u32 type, vector<u8> packet)
            { ProcessPacketsOnServer(client, type, std::move(packet)); };
            while (ProccessNetworkData(client.recvb, interpret))
            {
            }
        }
    }

    void operator()()
    {
        try
        {
            Loop();
        }
        catch (const std::exception &)
        {
            client.socket->close();
        }
    }
};

struct ParentImpl
{
    Store fileStore;
    int m_portOffset;
    std::thread listenerLoop;
    std::atomic<bool> appActive{true};

    void ParentLoop()
    {
        vector<std::future<void>> clients;
        int id{};

        try
        {
            de::SocketAddress addr;
            addr.setHost("localhost");
            addr.setPort(DefaultPortIPC + m_portOffset);
            de::Socket listener;
            listener.listen(addr);

            while (appActive)
            {
                remove_erase_if(clients, [](const std::future<void> &c) { return is_ready(c); });
                ChildConnection client{++id, std::unique_ptr<de::Socket>(listener.accept()), appActive, vector<u8>{}};
                clients.push_back(CreateClientThread(std::move(client)));
            }
        }
        catch (const std::exception &)
        {
            appActive = false;
        }
    }

    ParentImpl(const int portOffset) : m_portOffset{portOffset}, listenerLoop{[this]() { ParentLoop(); }}
    {
    }

    ~ParentImpl()
    {
        appActive = false;

        // Dummy connection to trigger accept()
        de::SocketAddress addr;
        addr.setHost("localhost");
        addr.setPort(DefaultPortIPC + m_portOffset);
        de::Socket socket;

        try
        {
            socket.connect(addr);
        }
        catch (const de::SocketError &)
        {
        }

        try
        {
            socket.close();
        }
        catch (const de::SocketError &)
        {
        }

        listenerLoop.join();
    }

    std::future<void> CreateClientThread(ChildConnection client)
    {
        return std::async(PacketsLoop{std::move(client), fileStore});
    }
};

Parent::Parent(const int portOffset)
{
    impl.reset(new ParentImpl(portOffset));
}

Parent::~Parent()
{
}

bool Parent::SetFile(const string &name, const std::vector<u8> &content)
{
    return impl->fileStore.Set(name, content);
}

vector<u8> Parent::GetFile(const string &name)
{
    vector<u8> content;
    bool result = impl->fileStore.Get(name, content, false);
    if (result)
        return content;
    else
        return {};
}

struct ChildImpl
{
    ChildImpl(const int portOffset) : m_portOffset(portOffset)
    {
        connection.reset(new Server{"localhost:" + std::to_string(DefaultPortIPC + m_portOffset)});
    }

    int m_portOffset;
    std::unique_ptr<Server> connection;
};

Child::Child(const int portOffset)
{
    impl.reset(new ChildImpl(portOffset));
}

Child::~Child()
{
}

bool Child::SetFile(const string &name, const std::vector<u8> &content)
{
    StoreContentRequest request;
    request.name = name;
    request.data = content;
    StoreContentResponse response;
    impl->connection->SendRequest(request, response);
    return response.status;
}

std::vector<u8> Child::GetFile(const string &name)
{
    GetContentRequest request;
    request.path         = name;
    request.physicalFile = false;
    request.removeAfter  = false;
    GetContentResponse response;
    impl->connection->SendRequest(request, response);
    if (response.status == true)
        return response.data;
    else
        return {};
}

} // namespace ipc

} // namespace vksc_server