hdfsEngine.cpp

#include <iostream>
#include <algorithm>
#include <cmath>

#include <string>
#include <unordered_map>

#include "hdfs.h"
#include "hdfsEngine.h"

faster::hdfsEngine::hdfsEngine(){
    _ready = false;
    _fs = hdfsConnect("default", 0);
    _ready = true;
}
faster::hdfsEngine::~hdfsEngine(){
    hdfsDisconnect((hdfsFS) _fs);
}

bool faster::hdfsEngine::isReady(){
    return _ready;
}

bool faster::hdfsEngine::isConnected(){
    return (_fs != NULL);
}

faster::hdfsFile faster::hdfsEngine::open(std::string path, fileMode mode){
    return hdfsFile((hdfsFS) _fs, path, mode);
}

void faster::hdfsEngine::close(faster::hdfsFile & f){
    f.close();
}

void faster::hdfsEngine::del(std::string path){
    hdfsDelete((::hdfsFS)_fs, path.data(), 1);
}

bool faster::hdfsEngine::exists(std::string path){
    int r = hdfsExists((::hdfsFS) _fs, path.data());
    return (r == 0);
}

faster::hdfsFile::hdfsFile(void * fs, std::string & path, fileMode mode){
    _fs = fs;
    _path = path;
    _f = hdfsOpenFile((::hdfsFS) fs, path.data(), mode,0,0,0);
    _open = true;
    _buffer.resize(128*1024);
    _path = path;
}
faster::hdfsFile::~hdfsFile(){
    close();
}
void faster::hdfsFile::close(){
    if (_open){
        hdfsCloseFile((::hdfsFS) _fs, (::hdfsFile) _f);
        _open = false;
    }
}

size_t faster::hdfsFile::read(char * v, size_t n){
    return hdfsRead((::hdfsFS)_fs, (::hdfsFile)_f, (void*) v, n);
}

size_t faster::hdfsFile::write(char * v, size_t n){
    return hdfsWrite((::hdfsFS)_fs, (::hdfsFile)_f, (void*) v, n);
}

size_t faster::hdfsFile::seek(size_t offset){
    return hdfsSeek((::hdfsFS)_fs, (::hdfsFile)_f, (tOffset) offset);
}

size_t faster::hdfsFile::readLine(char * v, size_t n, char sep){
    char buffer[1024];
    int count = 0, m, k;
    char c ;

    do{
        k = 0;
        m = read(buffer, std::min(1024, (int) n - count));
        c = buffer[k];
        // Find the separator in this chunk
        while ((c != sep) && (k < m)){
            k++;
            c = buffer[k];
        }

        std::copy_n(buffer, k, &v[count]);
        count += k;
    }while (c != sep);

    // Go back to the end of the line
    int rest = m - k;
    seek(count - rest);

    return count;
}

std:: vector<std::deque<int>>  faster::hdfsFile::getBlocksLocations(){
    std::vector<std::deque<int>> loc;
    std::unordered_map<std::string,int> hostMap;
    //int i = 0;
    int hostsMapped = 0;

    hdfsFileInfo * info = hdfsGetPathInfo((::hdfsFS)_fs, _path.data());

    std::cerr << "Getting Blocks Locations for: " << _path << " (" << info->mSize << " bytes)";
    char *** blocksLoc = hdfsGetHosts((::hdfsFS)_fs, _path.data(), 0, info->mSize);
    int numBlocks = std::ceil((float)info->mSize/info->mBlockSize);
    loc.resize(numBlocks);

    std::cerr << ".\n";
    for ( int i = 0; i < numBlocks ; i++){
    //while ( blocksLoc[i] != NULL){
        int j = 0;
        char ** hostList = blocksLoc[i];
        std::cerr << "   (" << i << ") ";
        while ( hostList[j] != NULL ){
            std::cerr << hostList[j] << " ";
            auto search = hostMap.find(hostList[j]);
            if ( search == hostMap.end()){
                hostMap[hostList[j]] = hostsMapped ++;
            }
            auto & locList = loc[i];
            locList.insert(locList.end(), hostMap[hostList[j]]);
            j++;
        }
        //i++;
        std::cerr << "\n";
    }
    for ( int i = 0; i < numBlocks ; i++){
        std::cerr << "   (" << i << ") ";
        for (auto it = loc[i].begin() ; it != loc[i].end(); it++){
            std::cerr << *it << " ";
        }
        std::cerr << "\n";
    }


    hdfsFreeFileInfo(info,1);
    hdfsFreeHosts(blocksLoc);

    return loc;
}

void faster::hdfsFile::del(){
    close();
    hdfsDelete((::hdfsFS)_fs, _path.data(), 1);
}