ExampleFiFoQueue.cpp

#include <Base/Common/FiFoQueue.hh>
#include <iostream>

using namespace std;
using namespace BIAS;

pthread_mutex_t* mut;
int arr[4000000];

void* reader(void* arg) {
  FiFoQueue<int>* queue= (FiFoQueue<int>*)arg;
  int a;
  //fetch lots of data
  for (size_t i=0; i<1000000; i++) {
    queue->FetchSynced(a);
    //ensure fetched data was not fetched before
    BIASASSERT(arr[a]!=1);
    arr[a] = 1;
  }
  return NULL;
}

void* writer1(void* arg) {
  FiFoQueue<int>* queue= (FiFoQueue<int>*)arg;
  for (int i=0; i<1000000; i++) {
    queue->AppendSynced(i);
    //cout << "w1: " << i << endl << flush;
  }
  return NULL;
}

void* writer2(void* arg) {
  FiFoQueue<int>* queue= (FiFoQueue<int>*)arg;
  for (int i=1000000; i<2000000; i++) {
    queue->AppendSynced(i);
    //cout << "w2: " << i << endl << flush;
  }
  return NULL;
}

void* writer3(void* arg) {
  FiFoQueue<int>* queue= (FiFoQueue<int>*)arg;
  for (int i=2000000; i<3000000; i++) {
    queue->AppendSynced(i);
    //cout << "w2: " << i << endl << flush;
  }
  return NULL;
}

void* writer4(void* arg) {
  FiFoQueue<int>* queue= (FiFoQueue<int>*)arg;
  for (int i=3000000; i<4000000; i++) {
    queue->AppendSynced(i);
    //cout << "w2: " << i << endl << flush;
  }
  return NULL;
}


int main()
{
  //create a queue with 20 slots
  FiFoQueue<int>* queue = new FiFoQueue<int>(1);

  //debugging array, ensure that every data is read only once
  mut = new pthread_mutex_t();
  pthread_mutex_init(mut,NULL);
  for (int i=0; i<4000000; i++)
    arr[i] = 0;

  //create 4 reader and 4 writers
  pthread_t threadId1,threadId2,threadId3,threadId4,
    threadId5,threadId6,threadId7,threadId8;
  pthread_create(&threadId1,NULL,&reader,(void*)queue);
  pthread_create(&threadId2,NULL,&reader,(void*)queue);
  pthread_create(&threadId3,NULL,&reader,(void*)queue);
  pthread_create(&threadId4,NULL,&reader,(void*)queue);
  pthread_create(&threadId5,NULL,&writer1,(void*)queue);
  pthread_create(&threadId6,NULL,&writer2,(void*)queue);
  pthread_create(&threadId7,NULL,&writer3,(void*)queue);
  pthread_create(&threadId8,NULL,&writer4,(void*)queue);

  //wait for threads
  pthread_join(threadId1,NULL);
  pthread_join(threadId2,NULL);
  pthread_join(threadId3,NULL);
  pthread_join(threadId4,NULL);
  pthread_join(threadId5,NULL);
  pthread_join(threadId6,NULL);
  pthread_join(threadId7,NULL);
  pthread_join(threadId8,NULL);

  delete queue;

  return 0;
}