InputParser.h

Go to the documentation of this file.

/*
 * 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; either version 3 of the License, or
 * (at your option) any later version.
 *
 * Written (W) 2011 Shashwat Lal Das
 * Copyright (C) 2011 Berlin Institute of Technology and Max-Planck-Society
 */

#ifndef __INPUTPARSER_H__
#define __INPUTPARSER_H__

#include <shogun/lib/common.h>
#ifdef HAVE_PTHREAD

#include <shogun/io/SGIO.h>
#include <shogun/io/streaming/StreamingFile.h>
#include <shogun/io/streaming/ParseBuffer.h>
#include <pthread.h>

#define PARSER_DEFAULT_BUFFSIZE 100

namespace shogun
{
    enum E_EXAMPLE_TYPE
    {
        E_LABELLED = 1,
        E_UNLABELLED = 2
    };

template <class T> class CInputParser
{
public:

    CInputParser();

    ~CInputParser();

    void init(CStreamingFile* input_file, bool is_labelled = true, int32_t size = PARSER_DEFAULT_BUFFSIZE);

    bool is_running();

    int32_t get_number_of_features() { return number_of_features; }

    void set_read_vector(void (CStreamingFile::*func_ptr)(T* &vec, int32_t &len));

    void set_read_vector_and_label(void (CStreamingFile::*func_ptr)(T* &vec, int32_t &len, float64_t &label));

    int32_t get_vector_and_label(T* &feature_vector,
                     int32_t &length,
                     float64_t &label);

    int32_t get_vector_only(T* &feature_vector, int32_t &length);

    void set_free_vector_after_release(bool free_vec);

    void set_free_vectors_on_destruct(bool destroy);

    void start_parser();

    void* main_parse_loop(void* params);


    void copy_example_into_buffer(Example<T>* ex);

    Example<T>* retrieve_example();

    int32_t get_next_example(T* &feature_vector,
                 int32_t &length,
                 float64_t &label);

    int32_t get_next_example(T* &feature_vector,
                 int32_t &length);

    void finalize_example();

    void end_parser();

    void exit_parser();

    int32_t get_ring_size() { return ring_size; }

private:
    static void* parse_loop_entry_point(void* params);

public:
    bool parsing_done;  
    bool reading_done;  
    E_EXAMPLE_TYPE example_type; 
protected:
    void (CStreamingFile::*read_vector) (T* &vec, int32_t &len);

    void (CStreamingFile::*read_vector_and_label) (T* &vec, int32_t &len, float64_t &label);

    CStreamingFile* input_source;

    pthread_t parse_thread;

    CParseBuffer<T>* examples_ring;

    int32_t number_of_features;

    int32_t number_of_vectors_parsed;

    int32_t number_of_vectors_read;

    Example<T>* current_example;

    T* current_feature_vector;

    float64_t current_label;

    int32_t current_len;

    bool free_after_release;

    int32_t ring_size;

    pthread_mutex_t examples_state_lock;

    pthread_cond_t examples_state_changed;

};

template <class T>
    void CInputParser<T>::set_read_vector(void (CStreamingFile::*func_ptr)(T* &vec, int32_t &len))
{
    // Set read_vector to point to the function passed as arg
    read_vector=func_ptr;
}

template <class T>
    void CInputParser<T>::set_read_vector_and_label(void (CStreamingFile::*func_ptr)(T* &vec, int32_t &len, float64_t &label))
{
    // Set read_vector_and_label to point to the function passed as arg
    read_vector_and_label=func_ptr;
}

template <class T>
    CInputParser<T>::CInputParser()
{
    /* this line was commented out when I found it. However, the mutex locks
     * have to be initialised. Otherwise uninitialised memory error */
    //init(NULL, true, PARSER_DEFAULT_BUFFSIZE);
    pthread_mutex_init(&examples_state_lock, NULL);
    pthread_cond_init(&examples_state_changed, NULL);
    examples_ring=NULL;
    parsing_done=true;
    reading_done=true;
}

template <class T>
    CInputParser<T>::~CInputParser()
{
    pthread_mutex_destroy(&examples_state_lock);
    pthread_cond_destroy(&examples_state_changed);

    SG_UNREF(examples_ring);
}

template <class T>
    void CInputParser<T>::init(CStreamingFile* input_file, bool is_labelled, int32_t size)
{
    input_source = input_file;

    if (is_labelled == true)
        example_type = E_LABELLED;
    else
        example_type = E_UNLABELLED;

    examples_ring = new CParseBuffer<T>(size);
    SG_REF(examples_ring);

    parsing_done = false;
    reading_done = false;
    number_of_vectors_parsed = 0;
    number_of_vectors_read = 0;

    current_len = -1;
    current_label = -1;
    current_feature_vector = NULL;

    free_after_release=true;
    ring_size=size;
}

template <class T>
    void CInputParser<T>::set_free_vector_after_release(bool free_vec)
{
    free_after_release=free_vec;
}

template <class T>
    void CInputParser<T>::set_free_vectors_on_destruct(bool destroy)
{
    examples_ring->set_free_vectors_on_destruct(destroy);
}

template <class T>
    void CInputParser<T>::start_parser()
{
    SG_SDEBUG("entering CInputParser::start_parser()\n")
    if (is_running())
    {
        SG_SERROR("Parser thread is already running! Multiple parse threads not supported.\n")
    }

    SG_SDEBUG("creating parse thread\n")
    pthread_create(&parse_thread, NULL, parse_loop_entry_point, this);

    SG_SDEBUG("leaving CInputParser::start_parser()\n")
}

template <class T>
    void* CInputParser<T>::parse_loop_entry_point(void* params)
{
    ((CInputParser *) params)->main_parse_loop(params);

    return NULL;
}

template <class T>
    bool CInputParser<T>::is_running()
{
    SG_SDEBUG("entering CInputParser::is_running()\n")
    bool ret;

    pthread_mutex_lock(&examples_state_lock);

    if (parsing_done)
        if (reading_done)
            ret = false;
        else
            ret = true;
    else
        ret = false;

    pthread_mutex_unlock(&examples_state_lock);

    SG_SDEBUG("leaving CInputParser::is_running(), returning %d\n", ret)
    return ret;
}

template <class T>
    int32_t CInputParser<T>::get_vector_and_label(T* &feature_vector,
                              int32_t &length,
                              float64_t &label)
{
    (input_source->*read_vector_and_label)(feature_vector, length, label);

    if (length < 1)
    {
        // Problem reading the example
        return 0;
    }

    return 1;
}

template <class T>
    int32_t CInputParser<T>::get_vector_only(T* &feature_vector,
                         int32_t &length)
{
    (input_source->*read_vector)(feature_vector, length);

    if (length < 1)
    {
        // Problem reading the example
        return 0;
    }

    return 1;
}

template <class T>
    void CInputParser<T>::copy_example_into_buffer(Example<T>* ex)
{
    examples_ring->copy_example(ex);
}

template <class T> void* CInputParser<T>::main_parse_loop(void* params)
{
    // Read the examples into current_* objects
    // Instead of allocating mem for new objects each time
#ifdef HAVE_PTHREAD
    CInputParser* this_obj = (CInputParser *) params;
    this->input_source = this_obj->input_source;

    while (1)
    {
        pthread_mutex_lock(&examples_state_lock);
        if (parsing_done)
        {
            pthread_mutex_unlock(&examples_state_lock);
            return NULL;
        }
        pthread_mutex_unlock(&examples_state_lock);

        pthread_testcancel();

        current_example = examples_ring->get_free_example();
        current_feature_vector = current_example->fv;
        current_len = current_example->length;
        current_label = current_example->label;

        if (example_type == E_LABELLED)
            get_vector_and_label(current_feature_vector, current_len, current_label);
        else
            get_vector_only(current_feature_vector, current_len);

        if (current_len < 0)
        {
            pthread_mutex_lock(&examples_state_lock);
            parsing_done = true;
            pthread_cond_signal(&examples_state_changed);
            pthread_mutex_unlock(&examples_state_lock);
            return NULL;
        }

        current_example->label = current_label;
        current_example->fv = current_feature_vector;
        current_example->length = current_len;

        examples_ring->copy_example(current_example);

        pthread_mutex_lock(&examples_state_lock);
        number_of_vectors_parsed++;
        pthread_cond_signal(&examples_state_changed);
        pthread_mutex_unlock(&examples_state_lock);
    }
#endif /* HAVE_PTHREAD */
    return NULL;
}

template <class T> Example<T>* CInputParser<T>::retrieve_example()
{
    /* This function should be guarded by mutexes while calling  */
    Example<T> *ex;

    if (parsing_done)
    {
        if (number_of_vectors_read == number_of_vectors_parsed)
        {
            reading_done = true;
            /* Signal to waiting threads that no more examples are left */
            pthread_cond_signal(&examples_state_changed);
            return NULL;
        }
    }

    if (number_of_vectors_parsed <= 0)
        return NULL;

    if (number_of_vectors_read == number_of_vectors_parsed)
    {
        return NULL;
    }

    ex = examples_ring->get_unused_example();
    number_of_vectors_read++;

    return ex;
}

template <class T> int32_t CInputParser<T>::get_next_example(T* &fv,
        int32_t &length, float64_t &label)
{
    /* if reading is done, no more examples can be fetched. return 0
       else, if example can be read, get the example and return 1.
       otherwise, wait for further parsing, get the example and
       return 1 */

    Example<T> *ex;

    while (1)
    {
        if (reading_done)
            return 0;

        pthread_mutex_lock(&examples_state_lock);
        ex = retrieve_example();

        if (ex == NULL)
        {
            if (reading_done)
            {
                /* No more examples left, return */
                pthread_mutex_unlock(&examples_state_lock);
                return 0;
            }
            else
            {
                /* Examples left, wait for one to become ready */
                pthread_cond_wait(&examples_state_changed, &examples_state_lock);
                pthread_mutex_unlock(&examples_state_lock);
                continue;
            }
        }
        else
        {
            /* Example ready, return the example */
            pthread_mutex_unlock(&examples_state_lock);
            break;
        }
    }

    fv = ex->fv;
    length = ex->length;
    label = ex->label;

    return 1;
}

template <class T>
    int32_t CInputParser<T>::get_next_example(T* &fv, int32_t &length)
{
    float64_t label_dummy;

    return get_next_example(fv, length, label_dummy);
}

template <class T>
    void CInputParser<T>::finalize_example()
{
    examples_ring->finalize_example(free_after_release);
}

template <class T> void CInputParser<T>::end_parser()
{
    SG_SDEBUG("entering CInputParser::end_parser\n")
    SG_SDEBUG("joining parse thread\n")
    pthread_join(parse_thread, NULL);
    SG_SDEBUG("leaving CInputParser::end_parser\n")
}

template <class T> void CInputParser<T>::exit_parser()
{
    SG_SDEBUG("cancelling parse thread\n")
    pthread_cancel(parse_thread);
}
}

#endif /* HAVE_PTHREAD */

#endif // __INPUTPARSER_H__