<commit_before>/* BEGIN_COPYRIGHT
 *
 * Copyright 2009-2018 CRS4.
 *
 * 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.
 *
 * END_COPYRIGHT
 */

#include "hdfs_fs.h"
#include "hdfs_file.h"

#include <sstream>
#include <hdfs/hdfs.h>
#include <unicodeobject.h>
#include <errno.h>
#include <string.h>

#define MAX_WD_BUFFSIZE 2048

#define str_empty(s) ((s) == NULL || (*(s) == '\0'))

PyObject* FsClass_new(PyTypeObject *type, PyObject *args, PyObject *kwds)
{
    FsInfo *self;

    self = (FsInfo *)type->tp_alloc(type, 0);
    if (self != NULL) {

        self->host = NULL;
        self->port = 0;

        self->user = NULL;
        self->group = NULL;

        self->_fs = NULL;
    }

    return (PyObject *)self;
}


void FsClass_dealloc(FsInfo* self)
{
    Py_TYPE(self)->tp_free((PyObject*)self);
}


int FsClass_init(FsInfo *self, PyObject *args, PyObject *kwds)
{

    // XXX: This call to PyArg_ParseTuple doesn't support non-ASCII characters in
    // the input strings (host, user, group)
    if (! PyArg_ParseTuple(args, "z|izz",
            &(self->host), &(self->port),
            &(self->user), &(self->group)))
        return -1;

    if (str_empty(self->host))
        self->host = NULL;

    if (str_empty(self->user))
        self->user = NULL;

    if (str_empty(self->group))
        self->group = NULL;

    // Connect cycles and retries more than once if necessary.  Better let
    // other Python threads through.
    Py_BEGIN_ALLOW_THREADS;
        if (self->user != NULL) {
            self->_fs = hdfsConnectAsUser(self->host, self->port, self->user);

        } else {
            self->_fs = hdfsConnect(self->host, self->port);
        }
    Py_END_ALLOW_THREADS;

    if (!self->_fs) {
        PyErr_SetFromErrno(PyExc_RuntimeError);
        return -1;
    }

    return 0;
}


PyObject* FsClass_close(FsInfo* self)
{
    hdfsDisconnect(self->_fs);
    Py_RETURN_NONE;
}


PyObject* FsClass_get_working_directory(FsInfo* self) {

    const size_t bufferSize = MAX_WD_BUFFSIZE;
    char *buffer = (char*)PyMem_Malloc(bufferSize);
    if (!buffer)
        return PyErr_NoMemory();

    if (hdfsGetWorkingDirectory(self->_fs, buffer, bufferSize) == NULL) {
        PyErr_SetString(PyExc_RuntimeError, "Cannot get working directory.");
        PyMem_Free(buffer);
        return NULL;
    }

    PyObject* result = PyUnicode_FromString(buffer);
    PyMem_Free(buffer);
    if (!result)
        return PyErr_NoMemory();

    return result;
}

PyObject* FsClass_get_path_info(FsInfo* self, PyObject *args, PyObject *kwds) {

    char* path = NULL;
    PyObject* retval = NULL;
    hdfsFileInfo* info;

    if (!PyArg_ParseTuple(args, "es", "utf-8",  &path)) {
        return NULL;
    }

    if (str_empty(path)) {
        PyMem_Free(path);
        PyErr_SetString(PyExc_ValueError, "Empty path");
        return NULL;
    }

    Py_BEGIN_ALLOW_THREADS;
        info = hdfsGetPathInfo(self->_fs, path);
    Py_END_ALLOW_THREADS;
    if (info == NULL) {
        PyMem_Free(path);
        return PyErr_SetFromErrno(PyExc_IOError);
    }

    retval =
        Py_BuildValue("{s:O,s:s,s:s,s:i,s:i,s:h,s:s,s:h,s:i,s:O,s:L}",
            "name", PyUnicode_FromString(info->mName),
            "kind", info->mKind == kObjectKindDirectory ? "directory" : "file",
            "group", info->mGroup,
            "last_mod", info->mLastMod,
            "last_access", info->mLastAccess,
            "replication", info->mReplication,
            "owner", info->mOwner,
            "permissions", info->mPermissions,
            "block_size", info->mBlockSize,
            "path", PyUnicode_FromString(info->mName),
            "size", info->mSize
    );
    PyMem_Free(path);
    hdfsFreeFileInfo(info, 1);
    return retval;
}


PyObject* FsClass_get_hosts(FsInfo* self, PyObject *args, PyObject *kwds) {

    Py_ssize_t start, length;
    PyObject* result = NULL;
    char* path = NULL;
    char*** hosts = NULL;

    if (!PyArg_ParseTuple(args, "esnn", "utf-8", &path, &start, &length)) {
        return NULL;
    }

    if (str_empty(path)) {
        PyMem_Free(path);
        PyErr_SetString(PyExc_ValueError, "Empty path");
        return NULL;
    }

    if (start < 0 || length < 0) {
        PyMem_Free(path);
        PyErr_SetString(PyExc_ValueError,
                        "Start position and length must be >= 0");
        return NULL;
    }

    Py_BEGIN_ALLOW_THREADS;
        hosts = hdfsGetHosts(self->_fs, path, start, length);
    Py_END_ALLOW_THREADS;
    PyMem_Free(path);
    if (!hosts) {
        return PyErr_SetFromErrno(PyExc_RuntimeError);
    }

    result = PyList_New(0);
    if (!result) goto mem_error;

    for (int blockNumber = 0; hosts[blockNumber] != NULL; ++blockNumber)
    {
        PyObject* blockHosts = PyList_New(0);
        if (!blockHosts) goto mem_error;

        for (int iBlockHost = 0; hosts[blockNumber][iBlockHost] != NULL; ++iBlockHost)
        {
            PyObject* str = PyUnicode_FromString(hosts[blockNumber][iBlockHost]);
            if (!str) goto mem_error;
            if (PyList_Append(blockHosts, str) < 0) goto mem_error;
        }

        if (PyList_Append(result, blockHosts) < 0) goto mem_error;
    }
    goto done; // skip the mem_error section

mem_error:
    PyErr_SetString(PyExc_MemoryError, "Error allocating host structure");
    Py_XDECREF(result);
    result = NULL;
    // fall through
done:
    if (hosts) hdfsFreeHosts(hosts);
    return result;
}

PyObject* FsClass_get_default_block_size(FsInfo* self) {
    tOffset size = hdfsGetDefaultBlockSize(self->_fs);
    return PyLong_FromSsize_t(size);
}

PyObject* FsClass_get_used(FsInfo* self) {

    tOffset size = hdfsGetUsed(self->_fs);
    return PyLong_FromSsize_t(size);
}

PyObject* FsClass_set_replication(FsInfo* self, PyObject* args, PyObject* kwds) {

    char* path = NULL;
    short replication;
    int result;

    if (!PyArg_ParseTuple(args, "esh", "utf-8", &path, &replication))
        return NULL;

    if (str_empty(path)) {
        PyMem_Free(path);
        PyErr_SetString(PyExc_ValueError, "Empty path");
        return NULL;
    }

    Py_BEGIN_ALLOW_THREADS;
        result = hdfsSetReplication(self->_fs, path, replication);
    Py_END_ALLOW_THREADS;
    PyMem_Free(path);
    if (result < 0) {
	return PyErr_SetFromErrno(PyExc_IOError);
    }
    Py_RETURN_NONE;
}


PyObject* FsClass_set_working_directory(FsInfo* self, PyObject* args, PyObject* kwds) {

    char* path = NULL;
    int result;

    if (!PyArg_ParseTuple(args, "es", "utf-8", &path))
        return NULL;

    if (str_empty(path)) {
        PyMem_Free(path);
        PyErr_SetString(PyExc_ValueError, "Empty path");
        return NULL;
    }

    Py_BEGIN_ALLOW_THREADS;
        result = hdfsSetWorkingDirectory(self->_fs, path);
    Py_END_ALLOW_THREADS;
    PyMem_Free(path);
    if (result < 0) {
	return PyErr_SetFromErrno(PyExc_IOError);
    }
    Py_RETURN_NONE;
}


PyObject* FsClass_open_file(FsInfo* self, PyObject *args, PyObject *kwds)
{
    PyObject* retval = NULL;
    char* path = NULL;
    const char* mode = MODE_READ;
    int flags, buff_size, blocksize;
    short replication;
    hdfsFile file;
    tOffset size = 0;
    hdfsFileInfo* info;

    if (!PyArg_ParseTuple(args, "es|sihi",
                          "utf-8", &path, &mode, &buff_size, &replication,
                          &blocksize)) {
        return NULL;
    }

    if (str_empty(path)) {
        PyMem_Free(path);
        PyErr_SetString(PyExc_ValueError, "Empty path");
        return NULL;
    }

    if (strcmp(mode, MODE_READ) == 0) {
        flags = O_RDONLY;
    } else if (strcmp(mode, MODE_WRITE) == 0) {
        flags = O_WRONLY;
    } else if (strcmp(mode, MODE_APPEND) == 0) {
        flags = O_WRONLY | O_APPEND;
    } else {
        PyMem_Free(path);
        PyErr_SetString(PyExc_ValueError, "Invalid mode");
        return NULL;
    }<commit_msg>fix var init in CoreHdfsFs.open_file<commit_after>/* BEGIN_COPYRIGHT
 *
 * Copyright 2009-2018 CRS4.
 *
 * 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.
 *
 * END_COPYRIGHT
 */

#include "hdfs_fs.h"
#include "hdfs_file.h"

#include <sstream>
#include <hdfs/hdfs.h>
#include <unicodeobject.h>
#include <errno.h>
#include <string.h>

#define MAX_WD_BUFFSIZE 2048

#define str_empty(s) ((s) == NULL || (*(s) == '\0'))

PyObject* FsClass_new(PyTypeObject *type, PyObject *args, PyObject *kwds)
{
    FsInfo *self;

    self = (FsInfo *)type->tp_alloc(type, 0);
    if (self != NULL) {

        self->host = NULL;
        self->port = 0;

        self->user = NULL;
        self->group = NULL;

        self->_fs = NULL;
    }

    return (PyObject *)self;
}


void FsClass_dealloc(FsInfo* self)
{
    Py_TYPE(self)->tp_free((PyObject*)self);
}


int FsClass_init(FsInfo *self, PyObject *args, PyObject *kwds)
{

    // XXX: This call to PyArg_ParseTuple doesn't support non-ASCII characters in
    // the input strings (host, user, group)
    if (! PyArg_ParseTuple(args, "z|izz",
            &(self->host), &(self->port),
            &(self->user), &(self->group)))
        return -1;

    if (str_empty(self->host))
        self->host = NULL;

    if (str_empty(self->user))
        self->user = NULL;

    if (str_empty(self->group))
        self->group = NULL;

    // Connect cycles and retries more than once if necessary.  Better let
    // other Python threads through.
    Py_BEGIN_ALLOW_THREADS;
        if (self->user != NULL) {
            self->_fs = hdfsConnectAsUser(self->host, self->port, self->user);

        } else {
            self->_fs = hdfsConnect(self->host, self->port);
        }
    Py_END_ALLOW_THREADS;

    if (!self->_fs) {
        PyErr_SetFromErrno(PyExc_RuntimeError);
        return -1;
    }

    return 0;
}


PyObject* FsClass_close(FsInfo* self)
{
    hdfsDisconnect(self->_fs);
    Py_RETURN_NONE;
}


PyObject* FsClass_get_working_directory(FsInfo* self) {

    const size_t bufferSize = MAX_WD_BUFFSIZE;
    char *buffer = (char*)PyMem_Malloc(bufferSize);
    if (!buffer)
        return PyErr_NoMemory();

    if (hdfsGetWorkingDirectory(self->_fs, buffer, bufferSize) == NULL) {
        PyErr_SetString(PyExc_RuntimeError, "Cannot get working directory.");
        PyMem_Free(buffer);
        return NULL;
    }

    PyObject* result = PyUnicode_FromString(buffer);
    PyMem_Free(buffer);
    if (!result)
        return PyErr_NoMemory();

    return result;
}

PyObject* FsClass_get_path_info(FsInfo* self, PyObject *args, PyObject *kwds) {

    char* path = NULL;
    PyObject* retval = NULL;
    hdfsFileInfo* info;

    if (!PyArg_ParseTuple(args, "es", "utf-8",  &path)) {
        return NULL;
    }

    if (str_empty(path)) {
        PyMem_Free(path);
        PyErr_SetString(PyExc_ValueError, "Empty path");
        return NULL;
    }

    Py_BEGIN_ALLOW_THREADS;
        info = hdfsGetPathInfo(self->_fs, path);
    Py_END_ALLOW_THREADS;
    if (info == NULL) {
        PyMem_Free(path);
        return PyErr_SetFromErrno(PyExc_IOError);
    }

    retval =
        Py_BuildValue("{s:O,s:s,s:s,s:i,s:i,s:h,s:s,s:h,s:i,s:O,s:L}",
            "name", PyUnicode_FromString(info->mName),
            "kind", info->mKind == kObjectKindDirectory ? "directory" : "file",
            "group", info->mGroup,
            "last_mod", info->mLastMod,
            "last_access", info->mLastAccess,
            "replication", info->mReplication,
            "owner", info->mOwner,
            "permissions", info->mPermissions,
            "block_size", info->mBlockSize,
            "path", PyUnicode_FromString(info->mName),
            "size", info->mSize
    );
    PyMem_Free(path);
    hdfsFreeFileInfo(info, 1);
    return retval;
}


PyObject* FsClass_get_hosts(FsInfo* self, PyObject *args, PyObject *kwds) {

    Py_ssize_t start, length;
    PyObject* result = NULL;
    char* path = NULL;
    char*** hosts = NULL;

    if (!PyArg_ParseTuple(args, "esnn", "utf-8", &path, &start, &length)) {
        return NULL;
    }

    if (str_empty(path)) {
        PyMem_Free(path);
        PyErr_SetString(PyExc_ValueError, "Empty path");
        return NULL;
    }

    if (start < 0 || length < 0) {
        PyMem_Free(path);
        PyErr_SetString(PyExc_ValueError,
                        "Start position and length must be >= 0");
        return NULL;
    }

    Py_BEGIN_ALLOW_THREADS;
        hosts = hdfsGetHosts(self->_fs, path, start, length);
    Py_END_ALLOW_THREADS;
    PyMem_Free(path);
    if (!hosts) {
        return PyErr_SetFromErrno(PyExc_RuntimeError);
    }

    result = PyList_New(0);
    if (!result) goto mem_error;

    for (int blockNumber = 0; hosts[blockNumber] != NULL; ++blockNumber)
    {
        PyObject* blockHosts = PyList_New(0);
        if (!blockHosts) goto mem_error;

        for (int iBlockHost = 0; hosts[blockNumber][iBlockHost] != NULL; ++iBlockHost)
        {
            PyObject* str = PyUnicode_FromString(hosts[blockNumber][iBlockHost]);
            if (!str) goto mem_error;
            if (PyList_Append(blockHosts, str) < 0) goto mem_error;
        }

        if (PyList_Append(result, blockHosts) < 0) goto mem_error;
    }
    goto done; // skip the mem_error section

mem_error:
    PyErr_SetString(PyExc_MemoryError, "Error allocating host structure");
    Py_XDECREF(result);
    result = NULL;
    // fall through
done:
    if (hosts) hdfsFreeHosts(hosts);
    return result;
}

PyObject* FsClass_get_default_block_size(FsInfo* self) {
    tOffset size = hdfsGetDefaultBlockSize(self->_fs);
    return PyLong_FromSsize_t(size);
}

PyObject* FsClass_get_used(FsInfo* self) {

    tOffset size = hdfsGetUsed(self->_fs);
    return PyLong_FromSsize_t(size);
}

PyObject* FsClass_set_replication(FsInfo* self, PyObject* args, PyObject* kwds) {

    char* path = NULL;
    short replication;
    int result;

    if (!PyArg_ParseTuple(args, "esh", "utf-8", &path, &replication))
        return NULL;

    if (str_empty(path)) {
        PyMem_Free(path);
        PyErr_SetString(PyExc_ValueError, "Empty path");
        return NULL;
    }

    Py_BEGIN_ALLOW_THREADS;
        result = hdfsSetReplication(self->_fs, path, replication);
    Py_END_ALLOW_THREADS;
    PyMem_Free(path);
    if (result < 0) {
	return PyErr_SetFromErrno(PyExc_IOError);
    }
    Py_RETURN_NONE;
}


PyObject* FsClass_set_working_directory(FsInfo* self, PyObject* args, PyObject* kwds) {

    char* path = NULL;
    int result;

    if (!PyArg_ParseTuple(args, "es", "utf-8", &path))
        return NULL;

    if (str_empty(path)) {
        PyMem_Free(path);
        PyErr_SetString(PyExc_ValueError, "Empty path");
        return NULL;
    }

    Py_BEGIN_ALLOW_THREADS;
        result = hdfsSetWorkingDirectory(self->_fs, path);
    Py_END_ALLOW_THREADS;
    PyMem_Free(path);
    if (result < 0) {
	return PyErr_SetFromErrno(PyExc_IOError);
    }
    Py_RETURN_NONE;
}


PyObject* FsClass_open_file(FsInfo* self, PyObject *args, PyObject *kwds)
{
    PyObject* retval = NULL;
    char* path = NULL;
    const char* mode = MODE_READ;
    int flags = 0;
    int buff_size = 0;
    int blocksize = 0;
    short replication = 0;
    hdfsFile file = NULL;
    tOffset size = 0;
    hdfsFileInfo* info = NULL;

    if (!PyArg_ParseTuple(args, "es|sihi",
                          "utf-8", &path, &mode, &buff_size, &replication,
                          &blocksize)) {
        return NULL;
    }

    if (str_empty(path)) {
        PyMem_Free(path);
        PyErr_SetString(PyExc_ValueError, "Empty path");
        return NULL;
    }

    if (strcmp(mode, MODE_READ) == 0) {
        flags = O_RDONLY;
    } else if (strcmp(mode, MODE_WRITE) == 0) {
        flags = O_WRONLY;
    } else if (strcmp(mode, MODE_APPEND) == 0) {
        flags = O_WRONLY | O_APPEND;
    } else {
        PyMem_Free(path);
        PyErr_SetString(PyExc_ValueError, "Invalid mode");
        return NULL;
    }
{'fraction_non_alphanumeric': 0.12444365319565605, 'fraction_numerical': 0.0054596166399620205, 'mean_word_length': 2.741563055062167, 'pattern_counts': {'":': 0, '<': 25, '<?xml version=': 0, '>': 105, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '2623983', 'n_tokens_mistral': 5918, 'n_tokens_neox': 5757, 'n_words': 1298}