10 #ifndef __PARSEBUFFER_H__
11 #define __PARSEBUFFER_H__
24 enum E_IS_EXAMPLE_USED
67 template <
class T>
class CParseBuffer:
public CSGObject
75 CParseBuffer(int32_t size = 1024);
89 Example<T>* get_free_example()
91 pthread_mutex_lock(write_lock);
92 pthread_mutex_lock(&ex_in_use_mutex[ex_write_index]);
93 while (ex_used[ex_write_index] == E_NOT_USED)
94 pthread_cond_wait(&ex_in_use_cond[ex_write_index], &ex_in_use_mutex[ex_write_index]);
95 Example<T>* ex=&ex_ring[ex_write_index];
96 pthread_mutex_unlock(&ex_in_use_mutex[ex_write_index]);
97 pthread_mutex_unlock(write_lock);
110 int32_t write_example(Example<T>* ex);
117 Example<T>* return_example_to_read();
124 Example<T>* get_unused_example();
134 int32_t copy_example(Example<T>* ex);
143 void finalize_example(
bool free_after_release);
154 void set_free_vectors_on_destruct(
bool destroy)
156 free_vectors_on_destruct = destroy;
163 bool get_free_vectors_on_destruct()
165 return free_vectors_on_destruct;
173 virtual const char* get_name()
const {
return "ParseBuffer"; }
180 virtual void inc_read_index()
182 ex_read_index=(ex_read_index + 1) % ring_size;
189 virtual void inc_write_index()
191 ex_write_index=(ex_write_index + 1) % ring_size;
202 E_IS_EXAMPLE_USED* ex_used;
204 pthread_mutex_t* ex_in_use_mutex;
206 pthread_cond_t* ex_in_use_cond;
208 pthread_mutex_t* read_lock;
210 pthread_mutex_t* write_lock;
213 int32_t ex_write_index;
215 int32_t ex_read_index;
218 bool free_vectors_on_destruct;
221 template <
class T> CParseBuffer<T>::CParseBuffer(int32_t size)
224 ex_ring = SG_CALLOC(Example<T>, ring_size);
225 ex_used = SG_MALLOC(E_IS_EXAMPLE_USED, ring_size);
226 ex_in_use_mutex = SG_MALLOC(pthread_mutex_t, ring_size);
227 ex_in_use_cond = SG_MALLOC(pthread_cond_t, ring_size);
228 read_lock = SG_MALLOC(pthread_mutex_t, 1);
229 write_lock = SG_MALLOC(pthread_mutex_t, 1);
231 SG_SINFO(
"Initialized with ring size: %d.\n", ring_size)
236 for (int32_t i=0; i<ring_size; i++)
238 ex_used[i] = E_EMPTY;
244 ex_ring[i].label = FLT_MAX;
246 pthread_cond_init(&ex_in_use_cond[i], NULL);
247 pthread_mutex_init(&ex_in_use_mutex[i], NULL);
249 pthread_mutex_init(read_lock, NULL);
250 pthread_mutex_init(write_lock, NULL);
252 free_vectors_on_destruct =
true;
255 template <
class T> CParseBuffer<T>::~CParseBuffer()
257 for (int32_t i=0; i<ring_size; i++)
259 if (ex_ring[i].fv != NULL && free_vectors_on_destruct)
261 SG_DEBUG(
"%s::~%s(): destroying examples ring vector %d at %p\n",
262 get_name(), get_name(), i, ex_ring[i].fv);
263 SG_FREE(ex_ring[i].fv);
265 pthread_mutex_destroy(&ex_in_use_mutex[i]);
266 pthread_cond_destroy(&ex_in_use_cond[i]);
270 SG_FREE(ex_in_use_mutex);
271 SG_FREE(ex_in_use_cond);
278 int32_t CParseBuffer<T>::write_example(Example<T> *ex)
280 ex_ring[ex_write_index].label = ex->label;
281 ex_ring[ex_write_index].fv = ex->fv;
282 ex_ring[ex_write_index].length = ex->length;
283 ex_used[ex_write_index] = E_NOT_USED;
290 Example<T>* CParseBuffer<T>::return_example_to_read()
292 if (ex_read_index >= 0)
293 return &ex_ring[ex_read_index];
299 Example<T>* CParseBuffer<T>::get_unused_example()
301 pthread_mutex_lock(read_lock);
304 int32_t current_index = ex_read_index;
307 pthread_mutex_lock(&ex_in_use_mutex[current_index]);
309 if (ex_used[current_index] == E_NOT_USED)
310 ex = return_example_to_read();
314 pthread_mutex_unlock(&ex_in_use_mutex[current_index]);
316 pthread_mutex_unlock(read_lock);
321 int32_t CParseBuffer<T>::copy_example(Example<T> *ex)
323 pthread_mutex_lock(write_lock);
325 int32_t current_index = ex_write_index;
327 pthread_mutex_lock(&ex_in_use_mutex[current_index]);
328 while (ex_used[ex_write_index] == E_NOT_USED)
330 pthread_cond_wait(&ex_in_use_cond[ex_write_index], &ex_in_use_mutex[ex_write_index]);
333 ret = write_example(ex);
335 pthread_mutex_unlock(&ex_in_use_mutex[current_index]);
336 pthread_mutex_unlock(write_lock);
342 void CParseBuffer<T>::finalize_example(
bool free_after_release)
344 pthread_mutex_lock(read_lock);
345 pthread_mutex_lock(&ex_in_use_mutex[ex_read_index]);
346 ex_used[ex_read_index] = E_USED;
348 if (free_after_release)
350 SG_DEBUG(
"Freeing object in ring at index %d and address: %p.\n",
351 ex_read_index, ex_ring[ex_read_index].fv);
353 SG_FREE(ex_ring[ex_read_index].fv);
354 ex_ring[ex_read_index].fv=NULL;
357 pthread_cond_signal(&ex_in_use_cond[ex_read_index]);
358 pthread_mutex_unlock(&ex_in_use_mutex[ex_read_index]);
361 pthread_mutex_unlock(read_lock);
365 #endif // HAVE_PTHREAD
366 #endif // __PARSEBUFFER_H__