C++ BBQ Threading: Implementing Dynamic Speed Control

[sofanglom]

Here is the description of what I am working on. I wasn't able to figure it out during the semester but now that I have some time off during the break, I'd like to figure it out.

implement a bounded buffer queue (BBQ) , where lock variable is used to synchronize the access of queues. 20 threads are created with 10 producers and 10 consumers. The size of the queue MAX should be a constant and can be changed (at of time of testing). Your program should display the following events on screen:

Thread creation
Item ID # produced by thread number #
Item ID # consumed by thread number #
Waiting to produce by thread number #
Waiting to consume by thread number #
Your program should run with a command "project1 TP TC, where project1 is the executable program, TP represents the "initial" sleep time range limit for producing threads and TC represents the sleep time range limit for consuming threads.

Note: Producers and consumers can control their speed of producing/consuming by changing the sleeping time between two consecutive produce/consume operations. The sleeping interval t should be a random value within a range (0,T). For producers, it should dynamically change its speed of "producing" by changing its sleep time range TP as follows: it should gradually slow down the producing when the buffer is over 75% occupancy and stop producing (with condition variable) when buffer is 100% full. Similarly, it should gradually accelerate the producing when the buffer is below 25% occupancy and reaches twice the initial average speed when buffer is empty. For consumers, it should maintain the same range TC and stop consuming (with condition variable) when the buffer is empty.

Testing: Your program should run forever until the user quits it using CTRL + C in the command line or kill the process. Different combinations of (TP,TC) will be used to evaluate the correctness of your output.

Here is my code. My professor had us do it based off of figures in the textbook so we just had to add on to already existing code.

#include <iostream>
#include <stdio.h>
#include <thread> 
#include <cstdlib> 
#include <string>
#include <mutex> //prevents multiple access of shared data
#include <time.h>

using namespace std;

static const int producer_consumer = 10;  
static int MAX = 0;
static int id = 0; //production/consumption id
static int TP = 0; //for producing threads
static int TC = 0; //for consuming threads
static mutex itemLock; class TSQueue{
	std::mutex mutex
    public:
    TSQueue();
    TSQueue(int len);
    ~TSQueue(){};
    bool tryInsert(int item);
    bool tryRemove(int *item);

    private:
    int front;
    int nextEmpty;
    int items[];
};

TSQueue::TSQueue(){
    items[MAX];
    front = nextEmpty = 0;
}

bool TSQueue::tryInsert(int item){
    bool success = false;
    mutex.lock();
    if((nextEmpty - front) < MAX){
        items[nextEmpty % MAX] = item;
        nextEmpty++;
        success = true;
    }
    mutex.unlock();
    return success;
}
bool TSQueue::tryRemove(int *item){
    bool success = false;
    mutex.lock();
    if(front < nextEmpty){
        *item = items[front % MAX];
        front++;
        success = true;
    }
    mutex.unlock();
    return success;
}

static int genSleep(int t){
    return rand()%(t+1);
}

static int produceItem(){
    return rand()%100;
}

static TSQueue q;

void Producer(){
    TSQueue *queue = &q;
     std::thread::id this_id = std::this_thread::get_id();
  
    while(true){
        int queueItem = produceItem();
        itemLock.lock();
        bool success = queue->tryInsert(queueItem);
        if(success)
        itemLock.unlock();
        std::this_thread::sleep_for(std::chrono::seconds(genSleep(TP)));
    }
}void Consumer(){
    TSQueue *queue = &q;
    std::thread::id this_id = std::this_thread::get_id();
   
    while(true){
        int queueItem;
        itemLock.lock();
        bool success = queue->tryRemove(&queueItem);
        if(success)
        itemLock.unlock();
        std::this_thread::sleep_for(std::chrono::seconds(genSleep(TC)));
    }
}

void testInsert(){
    TSQueue *queue = &q;
    int i;
    for(i = 0; i < MAX; i++){
        if(queue->tryInsert(i))
            cout << "This " << i << " has been inserted" << endl;
    }
    
}

void testRemove(){
    TSQueue *queue = &q;
    int i, item;
    for(i = 0; i < MAX; i++){
        if(queue->tryRemove(&item))
            cout << "This " << item << " has been removed " << endl;
        else
            cout << "There are no items to be removed." << endl;
    }
}

int main(int argc, char *argv[])
{
    if(argc > 3 || arg < 3){
        cout << "Please enter 3 values.";
        return 0;
    }
    
    string str(argv[1]);
    TP = stoi(str);
    str = argv[2];
    TC = stoi(str);

    try{
        cout << "Set MAX" << endl;
        string t;
        cin >> t;
        MAX = stoi(t);
        cout << "setting MAX to: " << MAX << endl;
    }
    catch(const exception &e ){
        cout << "There is a MAX = 20" << endl;
        MAX = 20;
    }
   
    thread producers[producer_consumer];
    thread consumers[producer_consumer];
    int i;
    for(i = 0; i < producer_consumer; i++){
        producers[i] = std::thread(Producer);
        consumers[i] = std::thread(Consumer);
        std::this_thread::sleep_for(std::chrono::seconds(1));
    }

    for(i = 0; i < producer_consumer; i++){
        producers[i].join();
        consumers[i].join();
    }

    return 0;
}

What I could never figure out is how to implement that dynamically changing part for the producer and consumer.

 

[Future Bruno]

I attempted to do it in the Producer and Consumer functions by implementing if statements that would change the sleep time accordingly, but it never seemed to work.