fiphobo.cc

#include "fiphobo.h"

#include <cstdio>
#include <fcntl.h>
#include <iostream>
#include <stdio.h>
#include <stdlib.h>
#include <sys/stat.h>
#include <sys/types.h>
#include <unistd.h>

#include "FiPhoExceptions.h"

namespace fiphoboserver {

FiPhobo::FiPhobo()
{
    memset(&descriptor, 0, sizeof(descriptor));
    descriptor.xd_attrs = {0};
}

void FiPhobo::setBucketName(std::string bucketName)
{
#ifdef DEBUG
    std::cout << "Phobos_Layer::setBucketName" << std::endl;
#endif
    int rc = pho_attr_set(&descriptor.xd_attrs, "bucket", bucketName.c_str());
    if (rc) {
        throw fiphoexceptions::PhobosException("pho_attr_set", rc);
    }
}

void FiPhobo::setMetaData(
    std::vector<std::pair<std::string, std::string>> metaData)
{
#ifdef DEBUG
    std::cout << "Phobos_Layer::setMetaData" << std::endl;
#endif
    std::for_each(
        metaData.begin(), metaData.end(),
        [&](std::pair<std::string, std::string> pair) {
            int rc = pho_attr_set(
                &descriptor.xd_attrs, pair.first.c_str(), pair.second.c_str());
            if (rc) {
                throw fiphoexceptions::PhobosException("pho_attr_set", rc);
            }
        });
}

void FiPhobo::startPutOperation(unsigned int size, std::string objectID)
{
#ifdef DEBUG
    std::cout << "Phobos_Layer::startPutOperation" << std::endl;
#endif
    startFIFO();

    char* unconstedObjectID = new char[objectID.length() + 1];
    strcpy(unconstedObjectID, objectID.c_str());

    descriptor.xd_op    = PHO_XFER_OP_PUT;
    descriptor.xd_objid = unconstedObjectID;
    descriptor.xd_fd    = fifo_descriptor;
    descriptor.xd_size  = size;

#ifdef DEBUG
    std::cout << "Starting async phobos_put_cpp" << std::endl;
#endif
    phobos_result = std::async([&]() {
        std::cout << "Phobos Thread id " << std::this_thread::get_id()
                  << std::endl;
        return phobos_put_cpp(&descriptor, 1, NULL, NULL);
    });
    std::cout << "Main Thread id " << std::this_thread::get_id() << std::endl;

    // TODOOOO
    // delete [] unconstedObjectID;
}


static long int fifoWriteTemp(
    std::unique_ptr<folly::IOBuf> buffer, int fifo_descriptor)
{
    std::cout << "IO Thread id " << std::this_thread::get_id() << std::endl;
    return write(fifo_descriptor, buffer->data(), buffer->length());
}

void FiPhobo::addDataToFIFO(std::unique_ptr<folly::IOBuf> buffer)
{
#ifdef DEBUG
    std::cout << "Phobos_Layer::addDataToFIFO" << std::endl;
#endif
    if (putRunning) {
        int ioRC = put_result.get();
        if (ioRC < 0) {
            throw fiphoexceptions::FIFOException("write", ioRC);
        }
    }
    putRunning = true;

    if (buffer) {
        put_result =
            std::async([fd = fifo_descriptor, buffer = std::move(buffer)] {
                return write(fd, buffer->data(), buffer->length());
            });
        // io_result.wait();
    }
}


void FiPhobo::finishPUT()
{
#ifdef DEBUG
    std::cout << "Phobos_Layer::finishPUT" << std::endl;
#endif
    int ioRC = put_result.get();
    if (ioRC < 0) {
        throw fiphoexceptions::FIFOException("write", ioRC);
    }

    int phobosRC = phobos_result.get();
    if (phobosRC) {
        throw fiphoexceptions::PhobosException("pho_put", phobosRC);
    }

    putRunning = false;
}

void FiPhobo::startGetOperation(std::string objectID)
{
#ifdef DEBUG
    std::cout << "Phobos_Layer::startGetOperation" << std::endl;
#endif
    startFIFO();

    char* unconstedObjectID = new char[objectID.length() + 1];
    strcpy(unconstedObjectID, objectID.c_str());

    descriptor.xd_op    = PHO_XFER_OP_GET;
    descriptor.xd_objid = unconstedObjectID;
    descriptor.xd_fd    = fifo_descriptor;

#ifdef DEBUG
    std::cout << "Starting async phobos_get_cpp" << std::endl;
#endif
    phobos_result = std::async(
        [&]() { return phobos_get_cpp(&descriptor, 1, NULL, NULL); });
}

std::unique_ptr<folly::IOBuf> FiPhobo::getDataFromFIFO(size_t size)
{
#ifdef DEBUG
    std::cout << "Phobos_Layer::getDataFromFIFO" << std::endl;
#endif

    get_result = std::async([&] {
        std::unique_ptr<folly::IOBuf> buffer = folly::IOBuf::create(size);
        int readRC = read(fifo_descriptor, buffer->writableData(), size);
        if (readRC < 0) {
            throw fiphoexceptions::FIFOException("read", readRC);
        }
        return std::move(buffer);
    });

    int phobosRC = phobos_result.get();
    if (phobosRC) {
        throw fiphoexceptions::PhobosException("pho_get", phobosRC);
    }

    std::unique_ptr<folly::IOBuf> buffer;
    try {
        buffer = std::move(get_result.get());
    }
    // TODO: Does this make any sense or is it ok if I don't catch this here but
    // let the caller catch it - what he has to do anyway?
    catch (fiphoexceptions::FIFOException& fifoException) {
        throw fifoException;
    }

    return std::move(buffer);
}

FiPhobo::~FiPhobo()
{
#ifdef DEBUG
    std::cout << "Phobos_Layer::~Phobos_Layer" << std::endl;
#endif
    // pho_attrs_free(&descriptor.xd_attrs); // TODO: Is that done in
    // desc_destroy??
    pho_xfer_desc_destroy_cpp(&descriptor);
    if (fifo_descriptor > 0) {
        close(fifo_descriptor);
    }
    if (access(fifoName.c_str(), F_OK) != -1) {
        remove(fifoName.c_str());
    }
}

void FiPhobo::startFIFO()
{
#ifdef DEBUG
    std::cout << "Phobos_Layer::startFIFO" << std::endl;
#endif
    int rc = mkfifo(fifoName.c_str(), 0777);
    if (rc < 0) {
        if (errno != EEXIST) {
            throw fiphoexceptions::FIFOException("mkfifo", errno);
            return;
        }
    }

    fifo_descriptor = open(fifoName.c_str(), O_RDWR);
    if (fifo_descriptor < 0) {
        std::stringstream ss;
        ss << "open with filename " << fifoName.c_str();
        throw fiphoexceptions::FIFOException(ss.str(), errno);
        return;
    }

#ifdef DEBUG
    std::cout << "End of Phobos_Layer::startFIFO" << std::endl;
#endif
}
}  // namespace fiphoboserver