<commit_before>// Copyright <NAME>, SwampBits LLC 2014
// BSD License

#include <cstdio>
#include <cstdlib>

#include "ThreadPoolQueue.h"
#include "ThreadingFactory.h"
#include "ConditionVariable.h"
#include "MutexLock.h"
#include "Logger.h"
#include "BasicException.h"

using namespace chaudiere;

//******************************************************************************

ThreadPoolQueue::ThreadPoolQueue(ThreadingFactory* threadingFactory) :
   m_threadingFactory(threadingFactory),
   m_mutex(NULL),
   m_condQueueNotEmpty(NULL),
   m_isInitialized(false),
   m_isRunning(false) {

   Logger::logInstanceCreate("ThreadPoolQueue");

   try {
      m_mutex = m_threadingFactory->createMutex("ThreadPoolQueue");
      m_condQueueNotEmpty = m_threadingFactory->createConditionVariable("queue-not-empty");
         
      if ((m_mutex != NULL) &&
          (m_condQueueNotEmpty != NULL)) {
         m_isInitialized = true;
         m_isRunning = true;
      } else {
         Logger::error("unable to initialize ThreadPoolQueue");
         if (NULL == m_mutex) {
            Logger::error("unable to create mutex");
         }
         if (NULL == m_condQueueNotEmpty) {
            Logger::error("unable to create queue not empty condition variable");
         }
         printf("error: unable to initialize thread pool queue, aborting\n");
         exit(1);
      }
   } catch (const BasicException& be) {
      Logger::error("exception setting up thread pool queue: " + be.whatString());
   } catch (const std::exception& e) {
      Logger::error("exception setting up thread pool queue: " + std::string(e.what()));
   } catch (...) {
      Logger::error("unknown exception setting up thread pool queue");
   }
}

//******************************************************************************

ThreadPoolQueue::~ThreadPoolQueue() {
   Logger::logInstanceDestroy("ThreadPoolQueue");
   m_isRunning = false;
   if (NULL != m_mutex) {
      delete m_mutex;
   }
   if (NULL != m_condQueueNotEmpty) {
      delete m_condQueueNotEmpty;
   }
}

//******************************************************************************

bool ThreadPoolQueue::addRequest(Runnable* runnableRequest) {
   if (!m_isInitialized) {
      Logger::log(Warning, "ThreadPoolQueue::addRequest queue not initialized");
      return false;
   }

   if (!runnableRequest) {
      Logger::log(Warning, "ThreadPoolQueue::addRequest rejecting NULL request");
      return false;
   }
  
   MutexLock lock(*m_mutex, "ThreadPoolQueue::addRequest");
   
   if (!m_isRunning) {
      Logger::log(Warning, "ThreadPoolQueue::addRequest rejecting request, queue is shutting down");
      return false;
   }
   
   //if (!m_mutex->haveValidMutex()) {
   //   Logger::error("don't have valid mutex in addRequest");
   //   ::exit(1);
   //}

   Logger::log(Debug, "ThreadPoolQueue::addRequest accepting request");
   
   const bool wasEmpty = m_queue.empty();
   
   // add new request to the queue
   m_queue.push_back(runnableRequest);
   
   // did we just transition from QUEUE_EMPTY to QUEUE_NOT_EMPTY?
   if (wasEmpty) {
      // signal QUEUE_NOT_EMPTY (wake up a worker thread)
      Logger::log(Debug, "signalling queue_not_empty");
      m_condQueueNotEmpty->notifyAll();
   }
  
   return true;
}

//******************************************************************************

Runnable* ThreadPoolQueue::takeRequest() {
   if (!m_isInitialized) {
      Logger::log(Warning, "ThreadPoolQueue::takeRequest queue not initialized");
      return NULL;
   }

   MutexLock lock(*m_mutex, "ThreadPoolQueue::takeRequest");

   // is the queue shut down?
   if (!m_isRunning) {
      return NULL;
   }
   
   //if (!m_mutex->haveValidMutex()) {
   //   Logger::error("don't have valid mutex in takeRequest");
   //   exit(1);
   //}
   
   Runnable* request = NULL;
  
   // is the queue empty?
   while (m_queue.empty()) { // && m_isRunning) {
      // empty queue -- wait for QUEUE_NOT_EMPTY event
      m_condQueueNotEmpty->wait(m_mutex);
   }

   if (!m_queue.empty()) {
      // take a request from the queue
      request = m_queue.front();
      m_queue.pop_front();
   }

   return request;
}

//******************************************************************************

bool ThreadPoolQueue::startUp() {
   bool wasStarted = true;<commit_msg>fix return type of startUp<commit_after>// Copyright <NAME>, SwampBits LLC 2014
// BSD License

#include <cstdio>
#include <cstdlib>

#include "ThreadPoolQueue.h"
#include "ThreadingFactory.h"
#include "ConditionVariable.h"
#include "MutexLock.h"
#include "Logger.h"
#include "BasicException.h"

using namespace chaudiere;

//******************************************************************************

ThreadPoolQueue::ThreadPoolQueue(ThreadingFactory* threadingFactory) :
   m_threadingFactory(threadingFactory),
   m_mutex(NULL),
   m_condQueueNotEmpty(NULL),
   m_isInitialized(false),
   m_isRunning(false) {

   Logger::logInstanceCreate("ThreadPoolQueue");

   try {
      m_mutex = m_threadingFactory->createMutex("ThreadPoolQueue");
      m_condQueueNotEmpty = m_threadingFactory->createConditionVariable("queue-not-empty");
         
      if ((m_mutex != NULL) &&
          (m_condQueueNotEmpty != NULL)) {
         m_isInitialized = true;
         m_isRunning = true;
      } else {
         Logger::error("unable to initialize ThreadPoolQueue");
         if (NULL == m_mutex) {
            Logger::error("unable to create mutex");
         }
         if (NULL == m_condQueueNotEmpty) {
            Logger::error("unable to create queue not empty condition variable");
         }
         printf("error: unable to initialize thread pool queue, aborting\n");
         exit(1);
      }
   } catch (const BasicException& be) {
      Logger::error("exception setting up thread pool queue: " + be.whatString());
   } catch (const std::exception& e) {
      Logger::error("exception setting up thread pool queue: " + std::string(e.what()));
   } catch (...) {
      Logger::error("unknown exception setting up thread pool queue");
   }
}

//******************************************************************************

ThreadPoolQueue::~ThreadPoolQueue() {
   Logger::logInstanceDestroy("ThreadPoolQueue");
   m_isRunning = false;
   if (NULL != m_mutex) {
      delete m_mutex;
   }
   if (NULL != m_condQueueNotEmpty) {
      delete m_condQueueNotEmpty;
   }
}

//******************************************************************************

bool ThreadPoolQueue::addRequest(Runnable* runnableRequest) {
   if (!m_isInitialized) {
      Logger::log(Warning, "ThreadPoolQueue::addRequest queue not initialized");
      return false;
   }

   if (!runnableRequest) {
      Logger::log(Warning, "ThreadPoolQueue::addRequest rejecting NULL request");
      return false;
   }
  
   MutexLock lock(*m_mutex, "ThreadPoolQueue::addRequest");
   
   if (!m_isRunning) {
      Logger::log(Warning, "ThreadPoolQueue::addRequest rejecting request, queue is shutting down");
      return false;
   }
   
   //if (!m_mutex->haveValidMutex()) {
   //   Logger::error("don't have valid mutex in addRequest");
   //   ::exit(1);
   //}

   Logger::log(Debug, "ThreadPoolQueue::addRequest accepting request");
   
   const bool wasEmpty = m_queue.empty();
   
   // add new request to the queue
   m_queue.push_back(runnableRequest);
   
   // did we just transition from QUEUE_EMPTY to QUEUE_NOT_EMPTY?
   if (wasEmpty) {
      // signal QUEUE_NOT_EMPTY (wake up a worker thread)
      Logger::log(Debug, "signalling queue_not_empty");
      m_condQueueNotEmpty->notifyAll();
   }
  
   return true;
}

//******************************************************************************

Runnable* ThreadPoolQueue::takeRequest() {
   if (!m_isInitialized) {
      Logger::log(Warning, "ThreadPoolQueue::takeRequest queue not initialized");
      return NULL;
   }

   MutexLock lock(*m_mutex, "ThreadPoolQueue::takeRequest");

   // is the queue shut down?
   if (!m_isRunning) {
      return NULL;
   }
   
   //if (!m_mutex->haveValidMutex()) {
   //   Logger::error("don't have valid mutex in takeRequest");
   //   exit(1);
   //}
   
   Runnable* request = NULL;
  
   // is the queue empty?
   while (m_queue.empty()) { // && m_isRunning) {
      // empty queue -- wait for QUEUE_NOT_EMPTY event
      m_condQueueNotEmpty->wait(m_mutex);
   }

   if (!m_queue.empty()) {
      // take a request from the queue
      request = m_queue.front();
      m_queue.pop_front();
   }

   return request;
}

//******************************************************************************

bool ThreadPoolQueue::startUp() {
   bool wasStarted = false;
{'fraction_non_alphanumeric': 0.20825645756457564, 'fraction_numerical': 0.0016143911439114391, 'mean_word_length': 3.5289817232375977, 'pattern_counts': {'":': 0, '<': 9, '<?xml version=': 0, '>': 21, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '2937618', 'n_tokens_mistral': 2518, 'n_tokens_neox': 2462, 'n_words': 553}