source: ThirdParty/JobMarket/src/unittests/WorkerPoolUnitTest.cpp@ 4464ef

/*
 * Project: JobMarket
 * Description: asynchronous Server/Controller/Client-approach to parallel computing, based on boost::asio
 * Copyright (C)  2010 Frederik Heber. All rights reserved.
 * 
 */

/*
 * WorkerPoolUnitTest.cpp
 *
 *  Created on: Feb 28, 2012
 *      Author: heber
 */

// include config.h
#ifdef HAVE_CONFIG_H
#include <config.h>
#endif

#include <cppunit/CompilerOutputter.h>
#include <cppunit/extensions/TestFactoryRegistry.h>
#include <cppunit/ui/text/TestRunner.h>

#include "WorkerPoolUnitTest.hpp"

#include <vector>

#include "CodePatterns/Assert.hpp"
#include "CodePatterns/Observer/Channels.hpp"
#include "JobMarket/Pool/WorkerPool.hpp"
#include "JobMarket/WorkerAddress.hpp"


#ifdef HAVE_TESTRUNNER
#include "JobMarket/UnitTestMain.hpp"
#endif /*HAVE_TESTRUNNER*/

/********************************************** Test classes **************************************/

#include "stubs/FragmentJobStub.hpp"
#include "unittests/stubs/ObserverStub.hpp"

// Registers the fixture into the 'registry'
CPPUNIT_TEST_SUITE_REGISTRATION( WorkerPoolTest );


void WorkerPoolTest::setUp()
{
  // Throw assertions
  ASSERT_DO(Assert::Throw);

  pool = new WorkerPool();
  addobserver = new NotificationObserver(pool->getChannel(WorkerPool::WorkerAdded));
  idleobserver = new NotificationObserver(pool->getChannel(WorkerPool::WorkerIdle));

  // and sign on
  pool->signOn(addobserver, WorkerPool::WorkerAdded);
  pool->signOn(idleobserver, WorkerPool::WorkerIdle);
}

void WorkerPoolTest::tearDown()
{
  pool->signOff(addobserver, WorkerPool::WorkerAdded);
  pool->signOff(idleobserver, WorkerPool::WorkerIdle);

  delete addobserver;
  delete idleobserver;
  delete pool;
}

/** UnitTest for working add/removeWorker
 */
void WorkerPoolTest::WorkerTest()
{
  WorkerAddress address("service", "port");
  WorkerAddress otheraddress("otherservice", "otherport");

  // add worker
  CPPUNIT_ASSERT_EQUAL( (size_t)0, pool->idle_queue.size() );
  CPPUNIT_ASSERT_EQUAL( (size_t)0, pool->busy_queue.size() );
  CPPUNIT_ASSERT_EQUAL( (size_t)0, pool->pool.size() );
  CPPUNIT_ASSERT ( pool->begin_idle() == pool->end_idle() );
  CPPUNIT_ASSERT( pool->addWorker(address) );
  CPPUNIT_ASSERT_EQUAL( (size_t)1, pool->idle_queue.size() );
  CPPUNIT_ASSERT_EQUAL( (size_t)0, pool->busy_queue.size() );
  CPPUNIT_ASSERT_EQUAL( (size_t)1, pool->pool.size() );
  CPPUNIT_ASSERT ( pool->begin_idle() != pool->end_idle() );
  CPPUNIT_ASSERT_EQUAL( address, pool->begin_idle()->second );
  CPPUNIT_ASSERT( addobserver->wasNotified );
  CPPUNIT_ASSERT( idleobserver->wasNotified );
  addobserver->wasNotified = false;
  idleobserver->wasNotified = false;

  // remove non-present worker
  std::cout << "The following error message is intended and does not indicate a failure." << std::endl;
  CPPUNIT_ASSERT( !pool->removeWorker(otheraddress) );
  CPPUNIT_ASSERT( !addobserver->wasNotified );
  CPPUNIT_ASSERT( !idleobserver->wasNotified );

  // check for presence
  CPPUNIT_ASSERT( pool->presentInPool(address) );
  CPPUNIT_ASSERT( !pool->presentInPool(otheraddress) );

  // remove worker
  CPPUNIT_ASSERT( pool->removeWorker(address) );
  CPPUNIT_ASSERT_EQUAL( (size_t)0, pool->idle_queue.size() );
  CPPUNIT_ASSERT_EQUAL( (size_t)0, pool->busy_queue.size() );
  CPPUNIT_ASSERT_EQUAL( (size_t)0, pool->pool.size() );
  CPPUNIT_ASSERT( !addobserver->wasNotified );
  CPPUNIT_ASSERT( !idleobserver->wasNotified );
}

/** UnitTest for working (un)markWorkerBusy
 */
void WorkerPoolTest::markBusyTest()
{
  WorkerAddress address("service", "port");
  CPPUNIT_ASSERT( pool->addWorker(address) );

  // check that not busy
  CPPUNIT_ASSERT( !pool->isWorkerBusy(address) );
  CPPUNIT_ASSERT( !pool->hasBusyWorkers() );
  CPPUNIT_ASSERT_EQUAL( (size_t)1, pool->idle_queue.size() );
  CPPUNIT_ASSERT_EQUAL( (size_t)0, pool->busy_queue.size() );
  CPPUNIT_ASSERT_EQUAL( (size_t)1, pool->pool.size() );
  CPPUNIT_ASSERT( addobserver->wasNotified );
  CPPUNIT_ASSERT( idleobserver->wasNotified );
  addobserver->wasNotified = false;
  idleobserver->wasNotified = false;

  // unmark and check
  pool->unmarkWorkerBusy(address);
  CPPUNIT_ASSERT( !pool->isWorkerBusy(address) );
  CPPUNIT_ASSERT( !pool->hasBusyWorkers() );
  CPPUNIT_ASSERT_EQUAL( (size_t)1, pool->idle_queue.size() );
  CPPUNIT_ASSERT_EQUAL( (size_t)0, pool->busy_queue.size() );

  // mark as busy and check
  const WorkerAddress otheraddress = pool->getNextIdleWorker();
  CPPUNIT_ASSERT( pool->isWorkerBusy(address) );
  CPPUNIT_ASSERT_EQUAL( (size_t)0, pool->idle_queue.size() );
  CPPUNIT_ASSERT_EQUAL( (size_t)1, pool->busy_queue.size() );

  // get one more
  CPPUNIT_ASSERT( !pool->presentIdleWorkers() );
#ifndef NDEBUG
  std::cout << "The following assertion is intended and does not indicate a failure." << std::endl;
  CPPUNIT_ASSERT_THROW( pool->getNextIdleWorker(), Assert::AssertionFailure );
#else
  const WorkerAddress emptyAddress = pool->getNextIdleWorker();
#endif
  CPPUNIT_ASSERT( pool->isWorkerBusy(address) );
  CPPUNIT_ASSERT_EQUAL( (size_t)0, pool->idle_queue.size() );
  CPPUNIT_ASSERT_EQUAL( (size_t)1, pool->busy_queue.size() );
  CPPUNIT_ASSERT( !addobserver->wasNotified );
  CPPUNIT_ASSERT( !idleobserver->wasNotified );

  // unmark and check
  pool->unmarkWorkerBusy(address);
  CPPUNIT_ASSERT( !pool->isWorkerBusy(address) );
  CPPUNIT_ASSERT_EQUAL( (size_t)1, pool->idle_queue.size() );
  CPPUNIT_ASSERT_EQUAL( (size_t)0, pool->busy_queue.size() );
  CPPUNIT_ASSERT( !addobserver->wasNotified );
  CPPUNIT_ASSERT( idleobserver->wasNotified );
}