/**
	 * @brief	Invia la notifica di terminazione al thread
	 */
	void exitThread() {
		mutex.lock();
		busy = true;
		active = false;
		mutex.unlock();
		condition.notify_one();
	}

	void push(T const& data) {
		{
			boost::mutex::scoped_lock lock(mutex_);
			queue_.push(data);
		}
		condition_.notify_one();
	}

    void consumeToLog()
    {
        while (true)
        {
            try
            {
                typename LogObject::SmartPtr logObj;
                {
                    boost::mutex::scoped_lock lock(_qMutex);

                    while (_queue.size() == 0)
                    {
                        if (_isStopping)
                        {
                            std::cout << " Stopping consumeToLog Thread. " << std::endl;
                            return;
                        }
                        _qCondVar.wait(lock);
                    }

                    // Get the entry
                    logObj = _queue.front();
                    _queue.pop();
                }
                printLogObject(logObj);
            }
            catch (const boost::thread_interrupted& err)
            {
                std::cout << " Log::consumeToLog() - Got Interrupt Signal. " << _queue.size() << std::endl;
            }
        }//while
    }

 void log(LogLevel l, const std::string& msg)
 {
     typename StringLogObject::SmartPtr logObj(new StringLogObject(msg, l));
     boost::mutex::scoped_lock lock(_stringQMutex);
     _stringQ.push(logObj);
     _stringQCondVar.notify_one();
 }

    /* threat functions*/
    void datagen_thread_loop(void)
    {
        ostringstream testline;

        /* create testdata*/
        int jj=0;
        for(int ii=0; ii < 8*8-1; ii++) //64 = 63 chars + end line
        {
            if(++jj > 9) jj = 0;
            testline << jj;
        }
        testline << "\n";

        jj=0;
        while(record_thread_running && jj < NUM_LINES)
        {
    #if !defined(WITH_BOOST_TIME)
            clock_gettime(CLOCK_MONOTONIC, &ts_beg); // http://linux.die.net/man/3/clock_gettime
    #else
            start_time = boost::posix_time::microsec_clock::local_time();
    #endif

            // ********************************* //
            *active_buffer += testline.str().c_str();
            if(4*1024 < active_buffer->length()) //write 4 kbyte blocks
            {
                record_trigger.notify_all();
            }
            // ********************************* //

    #if !defined(WITH_BOOST_TIME)
            clock_gettime(CLOCK_MONOTONIC, &ts_end);
            v_fTime_s.push_back(ts_end.tv_sec);
            v_fTime_us.push_back(ts_end.tv_nsec/1e+3);
            v_fDifftime.push_back((ts_end.tv_sec - ts_beg.tv_sec)  + (ts_end.tv_nsec - ts_beg.tv_nsec) / 1e9);
            f_DT_s = (ts_end.tv_sec - ts_beg.tv_sec)  + (ts_end.tv_nsec - ts_beg.tv_nsec) / 1e9;
    #else
            stop_time = boost::posix_time::microsec_clock::local_time();
            time_duration = (stop_time - start_time);
            v_fTime_s.push_back( (stop_time-boost::posix_time::from_time_t(0)).total_seconds() );
            v_fTime_us.push_back( (stop_time-boost::posix_time::from_time_t(0)).fractional_seconds() );
            v_fDifftime.push_back( time_duration.total_seconds()+ time_duration.fractional_seconds()/ 1e6);
            f_DT_s = (time_duration.total_seconds()+ time_duration.fractional_seconds()/ 1e6);
    #endif
    #if defined(DEBUG)
            if(0.5 < 1.e+3*f_DT_s) // log only values above 0.5 ms
            {
                cout << "Line " << jj << " of " << NUM_LINES << ":"
                     << "\t" << v_fTime_s.back() << "." << v_fTime_us.back() << "s: "
                     << "\tdT: " << fixed << 1.e+3*f_DT_s << " ms"
                     << endl;
            }
    #endif
            boost::this_thread::sleep(boost::posix_time::microseconds(4*1e+6*f_DT_s)); //about 50% CPU load
            jj++;
        }//while

        datagen_done = true;
    }

 void operator()()
 {
     boost::mutex::scoped_lock lk(mut);
     while(!done)
     {
         cond.wait(lk);
     }
 }

 void wait()
 {
     boost::mutex::scoped_lock l(m);
     while(!flag)
     {
         cond.wait(l);
     }
 }

 void wait()
 {
     boost::unique_lock<boost::mutex> l(m);
     while(!flag)
     {
         cond.wait(l);
     }
 }

	 void Start(CThreadUnit** pTasks, int countTasks)
	 {
        {
            boost::lock_guard<boost::mutex> lock(m_Mut);
            m_bDataReady=true;
        }
        m_Cond.notify_one();
	 }

	//Entry point for pool threads.
	void execute()
	{
		while( true )
		{
			//Wait on condition variable while the task is empty and the pool is
			//still running.
			boost::unique_lock< boost::mutex > lock( mutex_ );
			while( tasks_.empty() && !stop_ )
			{
                cv_task_.wait( lock );
			}

			// Copy task locally and remove from the queue. This is done within
			// its own scope so that the task object is destructed immediately
			// after running the task. This is useful in the event that the function
			// contains shared_ptr arguments bound via bind.
			if( !tasks_.empty() )
			{
				++busy_;
				boost::function< void() > task = tasks_.front();
				tasks_.pop();

				lock.unlock();

				//Run the task.
				try
				{
					task();
				}
				catch( const std::exception& )
				{
					//Suppress all exceptions
				}

				// Task has finished, so increment count of available threads
				lock.lock();
				++processed_;
				++available_;
				--busy_;
				cv_finished_.notify_one();
			}
			else if( stop_ ) {
				break;
			}
        }
    }

	void step() {
		if(getRunningState() != RUNNING) {
			boost::lock_guard<boost::mutex> lock2(*esc64_mutex);

			esc64->step();
		}
		cv->notify_all();
	}

 void wait_and_pop ( Data& value )
 {
    boost::mutex::scoped_lock lock ( _mutex ) ;
    while ( _queue.empty () )
       _cond.wait ( lock ) ;
    value = _queue.front () ;
    _queue.pop () ;
 }

	unsigned long long waitAndGetFrontElement(T* &pFrontData) {
		boost::mutex::scoped_lock lock(_mutex);
		while (_queue.empty()) {
			_conditionVar.wait(lock);
		}
		pFrontData = _queue.front().data;
		return _queue.front().pushedTime;
	}

    // for message queue
    void push(T data)
    {
        boost::mutex::scoped_lock     lockHandle(mutexHandle);

        queueHandle.push(data);
        lockHandle.unlock();
        condFlag.notify_all();
    }

    bool wait_and_pop(T& data, boost::posix_time::microseconds waitTime)
    {
        boost::system_time            timeout = boost::get_system_time() + waitTime;
        boost::mutex::scoped_lock     lockHandle(mutexHandle);

        if (condFlag.timed_wait(lockHandle, timeout,
                                isEmpty(&queueHandle))) {
            data = queueHandle.front();
            queueHandle.pop();
            condFlag.notify_all();

            return true;
        }
        else {
            return false;
        }
    }

 void relative_timed_wait_with_predicate()
 {
     boost::unique_lock<boost::mutex> lock(mutex);
     if(cond_var.timed_wait(lock,boost::posix_time::seconds(timeout_seconds),check_flag(flag)) && flag)
     {
         ++woken;
     }
 }

 void operator()()
 {
     boost::unique_lock<boost::mutex> lk(mut);
     while(!done)
     {
         cond.wait(lk);
     }
 }

 virtual void moneyReceived(const string &txId, uint64_t amount)
 {
     std::cout << "wallet: " << wallet->mainAddress() << "**** just received money ("
               << txId  << ", " << wallet->displayAmount(amount) << ")" << std::endl;
     total_rx += amount;
     receive_triggered = true;
     cv_receive.notify_one();
 }

void Main::push(Command cmd) {
	boost::lock_guard<boost::mutex> lock(libcec_sync);
	if( running )
	{
		commands.push(cmd);
		libcec_cond.notify_one();
	}
}

	T pop()
	{
		boost::mutex::scoped_lock lock(_mutex);

		while(_queue.empty())
		{
			_conditionVar.wait(lock);
		}

		T result =_queue.front();
		_queue.pop();

		lock.unlock();
		_conditionVar.notify_one();

		return result;
	}