ParseBuffer.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 __PARSEBUFFER_H__
#define __PARSEBUFFER_H__

#include <shogun/lib/config.h>

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

#include <shogun/lib/DataType.h>
#include <pthread.h>

namespace shogun
{

enum E_IS_EXAMPLE_USED
{
    E_EMPTY = 1,
    E_NOT_USED = 2,
    E_USED = 3
};

template <class T>
class Example
{
public:
    float64_t label;
    T* fv;
    index_t length;
};

template <class T> class CParseBuffer: public CSGObject
{
public:
    CParseBuffer(int32_t size = 1024);

    ~CParseBuffer();

    Example<T>* get_free_example()
    {
        pthread_mutex_lock(write_lock);
        pthread_mutex_lock(&ex_in_use_mutex[ex_write_index]);
        while (ex_used[ex_write_index] == E_NOT_USED)
            pthread_cond_wait(&ex_in_use_cond[ex_write_index], &ex_in_use_mutex[ex_write_index]);
        Example<T>* ex=&ex_ring[ex_write_index];
        pthread_mutex_unlock(&ex_in_use_mutex[ex_write_index]);
        pthread_mutex_unlock(write_lock);

        return ex;
    }

    int32_t write_example(Example<T>* ex);

    Example<T>* return_example_to_read();

    Example<T>* get_unused_example();

    int32_t copy_example(Example<T>* ex);

    void finalize_example(bool free_after_release);

    void set_free_vectors_on_destruct(bool destroy)
    {
        free_vectors_on_destruct = destroy;
    }

    bool get_free_vectors_on_destruct()
    {
        return free_vectors_on_destruct;
    }

    virtual const char* get_name() const { return "ParseBuffer"; }

    void init_vector();

protected:
    virtual void inc_read_index()
    {
        ex_read_index=(ex_read_index + 1) % ring_size;
    }

    virtual void inc_write_index()
    {
        ex_write_index=(ex_write_index + 1) % ring_size;
    }

protected:

    int32_t ring_size;
    Example<T>* ex_ring;

    E_IS_EXAMPLE_USED* ex_used;
    pthread_mutex_t* ex_in_use_mutex;
    pthread_cond_t* ex_in_use_cond;
    pthread_mutex_t* read_lock;
    pthread_mutex_t* write_lock;

    int32_t ex_write_index;
    int32_t ex_read_index;

    bool free_vectors_on_destruct;
};


template <class T> void CParseBuffer<T>::init_vector()
{
    if (!free_vectors_on_destruct)
        return;
    for (int32_t i=0; i<ring_size; i++)
    {
        if(ex_ring[i].fv==NULL)
            ex_ring[i].fv = new T();
    }
}

template <class T> CParseBuffer<T>::CParseBuffer(int32_t size)
{
    ring_size = size;
    ex_ring = SG_CALLOC(Example<T>, ring_size);
    ex_used = SG_MALLOC(E_IS_EXAMPLE_USED, ring_size);
    ex_in_use_mutex = SG_MALLOC(pthread_mutex_t, ring_size);
    ex_in_use_cond = SG_MALLOC(pthread_cond_t, ring_size);
    read_lock = SG_MALLOC(pthread_mutex_t, 1);
    write_lock = SG_MALLOC(pthread_mutex_t, 1);

    SG_SINFO("Initialized with ring size: %d.\n", ring_size)

    ex_write_index = 0;
    ex_read_index = 0;

    for (int32_t i=0; i<ring_size; i++)
    {
        ex_used[i] = E_EMPTY;

        ex_ring[i].fv = NULL;
        ex_ring[i].length = 1;
        ex_ring[i].label = FLT_MAX;

        pthread_cond_init(&ex_in_use_cond[i], NULL);
        pthread_mutex_init(&ex_in_use_mutex[i], NULL);
    }
    pthread_mutex_init(read_lock, NULL);
    pthread_mutex_init(write_lock, NULL);

    free_vectors_on_destruct = true;
}

template <class T> CParseBuffer<T>::~CParseBuffer()
{
    for (int32_t i=0; i<ring_size; i++)
    {
        if (ex_ring[i].fv != NULL && free_vectors_on_destruct)
        {
            SG_DEBUG("%s::~%s(): destroying examples ring vector %d at %p\n",
                    get_name(), get_name(), i, ex_ring[i].fv);
            delete ex_ring[i].fv;
        }
        pthread_mutex_destroy(&ex_in_use_mutex[i]);
        pthread_cond_destroy(&ex_in_use_cond[i]);
    }
    SG_FREE(ex_ring);
    SG_FREE(ex_used);
    SG_FREE(ex_in_use_mutex);
    SG_FREE(ex_in_use_cond);

    SG_FREE(read_lock);
    SG_FREE(write_lock);
}

template <class T>
int32_t CParseBuffer<T>::write_example(Example<T> *ex)
{
    ex_ring[ex_write_index].label = ex->label;
    ex_ring[ex_write_index].fv = ex->fv;
    ex_ring[ex_write_index].length = ex->length;
    ex_used[ex_write_index] = E_NOT_USED;
    inc_write_index();

    return 1;
}

template <class T>
Example<T>* CParseBuffer<T>::return_example_to_read()
{
    if (ex_read_index >= 0)
        return &ex_ring[ex_read_index];
    else
        return NULL;
}

template <class T>
Example<T>* CParseBuffer<T>::get_unused_example()
{
    pthread_mutex_lock(read_lock);

    Example<T> *ex;
    int32_t current_index = ex_read_index;
    // Because read index will change after return_example_to_read

    pthread_mutex_lock(&ex_in_use_mutex[current_index]);

    if (ex_used[current_index] == E_NOT_USED)
        ex = return_example_to_read();
    else
        ex = NULL;

    pthread_mutex_unlock(&ex_in_use_mutex[current_index]);

    pthread_mutex_unlock(read_lock);
    return ex;
}

template <class T>
int32_t CParseBuffer<T>::copy_example(Example<T> *ex)
{
    pthread_mutex_lock(write_lock);
    int32_t ret;
    int32_t current_index = ex_write_index;

    pthread_mutex_lock(&ex_in_use_mutex[current_index]);
    while (ex_used[ex_write_index] == E_NOT_USED)
    {
        pthread_cond_wait(&ex_in_use_cond[ex_write_index], &ex_in_use_mutex[ex_write_index]);
    }

    ret = write_example(ex);

    pthread_mutex_unlock(&ex_in_use_mutex[current_index]);
    pthread_mutex_unlock(write_lock);

    return ret;
}

template <class T>
void CParseBuffer<T>::finalize_example(bool free_after_release)
{
    pthread_mutex_lock(read_lock);
    pthread_mutex_lock(&ex_in_use_mutex[ex_read_index]);
    ex_used[ex_read_index] = E_USED;

    if (free_after_release)
    {
        SG_DEBUG("Freeing object in ring at index %d and address: %p.\n",
             ex_read_index, ex_ring[ex_read_index].fv);

        SG_FREE(ex_ring[ex_read_index].fv);
        ex_ring[ex_read_index].fv=NULL;
    }

    pthread_cond_signal(&ex_in_use_cond[ex_read_index]);
    pthread_mutex_unlock(&ex_in_use_mutex[ex_read_index]);
    inc_read_index();

    pthread_mutex_unlock(read_lock);
}

}
#endif // HAVE_PTHREAD
#endif // __PARSEBUFFER_H__