<commit_before>/*
 * (C) Copyright 2014 Kurento (http://kurento.org/)
 *
 * All rights reserved. This program and the accompanying materials
 * are made available under the terms of the GNU Lesser General Public License
 * (LGPL) version 2.1 which accompanies this distribution, and is available at
 * http://www.gnu.org/licenses/lgpl-2.1.html
 *
 * This library is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
 * Lesser General Public License for more details.
 *
 */

#include <gst/gst.h>

#include "WorkerPool.hpp"
#include <atomic>

#define GST_CAT_DEFAULT kurento_worker_pool
GST_DEBUG_CATEGORY_STATIC (GST_CAT_DEFAULT);
#define GST_DEFAULT_NAME "KurentoWorkerPool"

const int WORKER_THREADS_TIMEOUT = 3; /* seconds */

namespace kurento
{

static void
workerThreadLoop ( boost::shared_ptr< boost::asio::io_service > io_service )
{
  bool running = true;

  while (running) {
    try {
      GST_DEBUG ("Working thread starting");
      io_service->run();
      running = false;
    } catch (std::exception &e) {
      GST_ERROR ("Unexpected error while running the server: %s", e.what() );
    } catch (...) {
      GST_ERROR ("Unexpected error while running the server");
    }
  }

  GST_DEBUG ("Working thread finished");
}

WorkerPool::WorkerPool (int threads)
{
  /* Prepare watcher */
  watcher_service = boost::shared_ptr< boost::asio::io_service >
                    ( new boost::asio::io_service () );
  watcher_work = std::shared_ptr< boost::asio::io_service::work >
                 ( new boost::asio::io_service::work (*watcher_service) );
  watcher = std::thread (std::bind (&workerThreadLoop, watcher_service) );

  /* Prepare pool of threads */
  io_service = boost::shared_ptr< boost::asio::io_service >
               ( new boost::asio::io_service () );
  work = std::shared_ptr< boost::asio::io_service::work >
         ( new boost::asio::io_service::work (*io_service) );

  for (int i = 0; i < threads; i++) {
    workers.push_back (std::thread (std::bind (&workerThreadLoop, io_service) ) );
  }
}

WorkerPool::~WorkerPool()
{
  // Executing queued tasks
  while (io_service->poll_one() ) {
  }

  watcher_service->stop();
  io_service->stop();

  watcher.join();

  for (uint i = 0; i < workers.size (); i++) {
    workers[i].join();
  }
}

static void
async_worker_test (std::shared_ptr<std::atomic<bool>> alive)
{
  *alive = true;
}

void
WorkerPool::checkWorkers ()
{
  std::shared_ptr<std::atomic<bool>> alive;<commit_msg>WorkerPool: Stop worker threads before execute pending tasks<commit_after>/*
 * (C) Copyright 2014 Kurento (http://kurento.org/)
 *
 * All rights reserved. This program and the accompanying materials
 * are made available under the terms of the GNU Lesser General Public License
 * (LGPL) version 2.1 which accompanies this distribution, and is available at
 * http://www.gnu.org/licenses/lgpl-2.1.html
 *
 * This library is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
 * Lesser General Public License for more details.
 *
 */

#include <gst/gst.h>

#include "WorkerPool.hpp"
#include <atomic>

#define GST_CAT_DEFAULT kurento_worker_pool
GST_DEBUG_CATEGORY_STATIC (GST_CAT_DEFAULT);
#define GST_DEFAULT_NAME "KurentoWorkerPool"

const int WORKER_THREADS_TIMEOUT = 3; /* seconds */

namespace kurento
{

static void
workerThreadLoop ( boost::shared_ptr< boost::asio::io_service > io_service )
{
  bool running = true;

  while (running) {
    try {
      GST_DEBUG ("Working thread starting");
      io_service->run();
      running = false;
    } catch (std::exception &e) {
      GST_ERROR ("Unexpected error while running the server: %s", e.what() );
    } catch (...) {
      GST_ERROR ("Unexpected error while running the server");
    }
  }

  GST_DEBUG ("Working thread finished");
}

WorkerPool::WorkerPool (int threads)
{
  /* Prepare watcher */
  watcher_service = boost::shared_ptr< boost::asio::io_service >
                    ( new boost::asio::io_service () );
  watcher_work = std::shared_ptr< boost::asio::io_service::work >
                 ( new boost::asio::io_service::work (*watcher_service) );
  watcher = std::thread (std::bind (&workerThreadLoop, watcher_service) );

  /* Prepare pool of threads */
  io_service = boost::shared_ptr< boost::asio::io_service >
               ( new boost::asio::io_service () );
  work = std::shared_ptr< boost::asio::io_service::work >
         ( new boost::asio::io_service::work (*io_service) );

  for (int i = 0; i < threads; i++) {
    workers.push_back (std::thread (std::bind (&workerThreadLoop, io_service) ) );
  }
}

WorkerPool::~WorkerPool()
{
  watcher_service->stop();
  io_service->stop();

  watcher.join();

  for (uint i = 0; i < workers.size (); i++) {
    workers[i].join();
  }

  // Executing queued tasks
  io_service->reset();

  while (io_service->poll() ) {
  }
}

static void
async_worker_test (std::shared_ptr<std::atomic<bool>> alive)
{
  *alive = true;
}

void
WorkerPool::checkWorkers ()
{
  std::shared_ptr<std::atomic<bool>> alive;
{'fraction_non_alphanumeric': 0.1348314606741573, 'fraction_numerical': 0.004189678156541611, 'mean_word_length': 3.8674698795180724, 'pattern_counts': {'":': 0, '<': 29, '<?xml version=': 0, '>': 34, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, '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': '3420372', 'n_tokens_mistral': 1703, 'n_tokens_neox': 1638, 'n_words': 445}