mariadb-columnstore-engine/utils/cloudio/SMComm.cpp

/* Copyright (C) 2019 MariaDB Corporaton

   This program is free software; you can redistribute it and/or
   modify it under the terms of the GNU General Public License
   as published by the Free Software Foundation; version 2 of
   the License.

   This program is distributed in the hope that it will be useful,
   but WITHOUT ANY WARRANTY; without even the implied warranty of
   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
   GNU General Public License for more details.

   You should have received a copy of the GNU General Public License
   along with this program; if not, write to the Free Software
   Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston,
   MA 02110-1301, USA. */

#include "SMComm.h"
#include "messageFormat.h"

using namespace std;
using namespace messageqcpp;

namespace
{
idbdatafile::SMComm *instance = NULL;
boost::mutex m;
};

namespace idbdatafile
{

SMComm * SMComm::get()
{
    if (instance)
        return instance;
        
    boost::mutex::scoped_lock sl(m);
    
    if (instance)
        return instance;
    instance = new SMComm();
    return instance;
}

// timesavers
#define common_exit(bs1, bs2, retCode) \
    { \
        buffers.returnByteStream(bs1); \
        buffers.returnByteStream(bs2); \
        return retCode; \
    }

// bs1 is the bytestream ptr with the command to SMComm.
// bs2 is the bytestream pointer with the response from SMComm.
// retCode is the var to store the return code in from the msg.
// returns with the output pointer at the fcn-specific data
#define check_for_error(bs1, bs2, retCode) \
    { \
        int l_errno; \
        *bs2 >> retCode; \
        if (retCode < 0) \
        { \
            *bs2 >> l_errno; \
            errno = l_errno; \
            common_exit(bs1, bs2, retCode); \
        } \
        else \
            errno = 0; \
    }
    
SMComm::SMComm()
{
    char buf[4096];
    cwd = ::getcwd(buf, 4096);
}

SMComm::~SMComm()
{
}

string SMComm::getAbsFilename(const string &filename)
{
    if (filename[0] == '/')
        return filename;
    else
        return cwd + '/' + filename;
}

int SMComm::open(const string &filename, const int mode, struct stat *statbuf)
{
    ByteStream *command = buffers.getByteStream();
    ByteStream *response = buffers.getByteStream();
    ssize_t err;
    string absfilename(getAbsFilename(filename));
    
    *command << (uint8_t) storagemanager::OPEN << mode << absfilename;
    err = sockets.send_recv(*command, response);
    if (err)
        common_exit(command, response, err);
        
    check_for_error(command, response, err);
    memcpy(statbuf, response->buf(), sizeof(*statbuf));
    common_exit(command, response, err);
}

ssize_t SMComm::pread(const string &filename, void *buf, const size_t count, const off_t offset)
{
    ByteStream *command = buffers.getByteStream();
    ByteStream *response = buffers.getByteStream();
    ssize_t err;
    string absfilename(getAbsFilename(filename));
    
    *command << (uint8_t) storagemanager::READ << count << offset << absfilename;
    err = sockets.send_recv(*command, response);
    if (err)
        common_exit(command, response, err);
    check_for_error(command, response, err);
    
    memcpy(buf, response->buf(), err);
    common_exit(command, response, err);
}

ssize_t SMComm::pwrite(const string &filename, const void *buf, const size_t count, const off_t offset)
{
    ByteStream *command = buffers.getByteStream();
    ByteStream *response = buffers.getByteStream();
    ssize_t err;
    string absfilename(getAbsFilename(filename));
    
    *command << (uint8_t) storagemanager::WRITE << count << offset << absfilename;
    command->needAtLeast(count);
    uint8_t *cmdBuf = command->getInputPtr();
    memcpy(cmdBuf, buf, count);
    command->advanceInputPtr(count);
    err = sockets.send_recv(*command, response);
    if (err)
        common_exit(command, response, err);
    check_for_error(command, response, err);
    common_exit(command, response, err);
}

ssize_t SMComm::append(const string &filename, const void *buf, const size_t count)
{
    ByteStream *command = buffers.getByteStream();
    ByteStream *response = buffers.getByteStream();
    ssize_t err;
    string absfilename(getAbsFilename(filename));

    *command << (uint8_t) storagemanager::APPEND << count << absfilename;
    command->needAtLeast(count);
    uint8_t *cmdBuf = command->getInputPtr();
    memcpy(cmdBuf, buf, count);
    command->advanceInputPtr(count);
    err = sockets.send_recv(*command, response);
    if (err)
        common_exit(command, response, err);
    check_for_error(command, response, err);
    common_exit(command, response, err);
}

int SMComm::unlink(const string &filename)
{
    ByteStream *command = buffers.getByteStream();
    ByteStream *response = buffers.getByteStream();
    ssize_t err;
    string absfilename(getAbsFilename(filename));
    
    *command << (uint8_t) storagemanager::UNLINK << absfilename;
    err = sockets.send_recv(*command, response);
    if (err)
        common_exit(command, response, err);
    check_for_error(command, response, err);
    common_exit(command, response, err);
}

int SMComm::stat(const string &filename, struct stat *statbuf)
{
    ByteStream *command = buffers.getByteStream();
    ByteStream *response = buffers.getByteStream();
    ssize_t err;
    string absfilename(getAbsFilename(filename));
    
    *command << (uint8_t) storagemanager::STAT << absfilename;
    err = sockets.send_recv(*command, response);
    if (err)
        common_exit(command, response, err);
    check_for_error(command, response, err);
    
    memcpy(statbuf, response->buf(), sizeof(*statbuf));
    common_exit(command, response, err);
}

int SMComm::truncate(const string &filename, const off64_t length)
{
    ByteStream *command = buffers.getByteStream();
    ByteStream *response = buffers.getByteStream();
    ssize_t err;
    string absfilename(getAbsFilename(filename));
    
    *command << (uint8_t) storagemanager::TRUNCATE << length << absfilename;
    err = sockets.send_recv(*command, response);
    if (err)
        common_exit(command, response, err);
    check_for_error(command, response, err);
    common_exit(command, response, err);
}

int SMComm::listDirectory(const string &path, list<string> *entries)
{
    ByteStream *command = buffers.getByteStream();
    ByteStream *response = buffers.getByteStream();
    ssize_t err;
    string abspath(getAbsFilename(path));
    
    *command << (uint8_t) storagemanager::LIST_DIRECTORY << abspath;
    err = sockets.send_recv(*command, response);
    if (err)
        common_exit(command, response, err);
    check_for_error(command, response, err);
    
    uint32_t numElements;
    string stmp;
    entries->clear();
    *response >> numElements;
    while (numElements > 0) {
        *response >> stmp;
        entries->push_back(stmp);
        numElements--;
    }
    common_exit(command, response, err);
}

int SMComm::ping()
{
    ByteStream *command = buffers.getByteStream();
    ByteStream *response = buffers.getByteStream();
    ssize_t err;
    
    *command << (uint8_t) storagemanager::PING;
    err = sockets.send_recv(*command, response);
    if (err)
        common_exit(command, response, err);
    check_for_error(command, response, err);
    common_exit(command, response, err);
}

int SMComm::copyFile(const string &file1, const string &file2)
{
    ByteStream *command = buffers.getByteStream();
    ByteStream *response = buffers.getByteStream();
    ssize_t err;
    string absfilename1(getAbsFilename(file1));
    string absfilename2(getAbsFilename(file2));
    
    *command << (uint8_t) storagemanager::COPY << absfilename1 << absfilename2;
    err = sockets.send_recv(*command, response);
    if (err)
        common_exit(command, response, err);
    check_for_error(command, response, err);
    common_exit(command, response, err);
}

}