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
 */

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

#ifndef __PARSEBUFFER_H__
#define __PARSEBUFFER_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;
    SGVector<T> fv;
};

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; }

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

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

    inline 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> 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 = new pthread_mutex_t;
    write_lock = new pthread_mutex_t;

    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.vector = new T();
        ex_ring[i].fv.vlen = 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.vector != NULL && free_vectors_on_destruct)
            delete ex_ring[i].fv.vector;
        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);

    delete read_lock;
    delete 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.vector = ex->fv.vector;
    ex_ring[ex_write_index].fv.vlen = ex->fv.vlen;
    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.vector);

        delete ex_ring[ex_read_index].fv.vector;
        ex_ring[ex_read_index].fv.vector=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 // __PARSEBUFFER_H__