path: root/cpp/src/IceStorm/SubscriberPool.cpp

// **********************************************************************
//
// Copyright (c) 2003-2006 ZeroC, Inc. All rights reserved.
//
// This copy of Ice is licensed to you under the terms described in the
// ICE_LICENSE file included in this distribution.
//
// **********************************************************************

#include <IceStorm/SubscriberPool.h>
#include <IceStorm/Instance.h>
#include <IceStorm/TraceLevels.h>
#include <IceStorm/Subscriber.h>

#include <Ice/Communicator.h>
#include <Ice/Properties.h>
#include <Ice/LoggerUtil.h>

using namespace IceStorm;
using namespace std;

namespace
{

class SubscriberPoolWorker : public IceUtil::Thread
{
public:

    SubscriberPoolWorker(const SubscriberPoolPtr& manager) :
	_manager(manager)
    {
	start();
    }

    ~SubscriberPoolWorker()
    {
    }

    virtual void
    run()
    {
	IceUtil::Time interval = IceUtil::Time::seconds(24 * 60); // A long time.
	SubscriberPtr sub;
	bool requeue = false;
	bool computeInterval = false;
	while(true)
	{
	    _manager->dequeue(sub, requeue, interval, computeInterval);
	    if(!sub)
	    {
		return;
	    }

	    //
	    // If SubscriberPool returns true then the subscriber
	    // needs to be SubscriberPooled again, so therefore we
	    // will re-enqueue the subscriber in the call to dequeue.
	    //
	    if(computeInterval)
	    {
		IceUtil::Time start = IceUtil::Time::now();
		requeue = sub->flush();
		interval = IceUtil::Time::now() - start;
	    }
	    else
	    {
		requeue = sub->flush();
		interval = IceUtil::Time::seconds(24 * 60); // A long time.
	    }
	}
    }

private:

    const SubscriberPoolPtr _manager;
};

}

SubscriberPoolMonitor::SubscriberPoolMonitor(const SubscriberPoolPtr& manager, const IceUtil::Time& timeout) :
    _manager(manager),
    _timeout(timeout),
    _needCheck(false),
    _destroyed(false)
{
    start();
}

SubscriberPoolMonitor::~SubscriberPoolMonitor()
{
}

void
SubscriberPoolMonitor::run()
{
    for(;;)
    {
        {
    	    Lock sync(*this);
    	    if(_destroyed)
    	    {
    	        return;
    	    }
    
    	    if(_needCheck)
    	    {
    	        timedWait(_timeout);
    	        //
    	        // Monitoring was stopped.
    	        //
    	        if(!_needCheck)
    	        {
    		    continue;
    	        }
		if(_destroyed)
		{
		    return;
		}
    	    }
    	    else
    	    {
    	        wait();
    	        continue;
    	    }
        }
        //
        // Call outside of the lock to prevent any deadlocks.
        //
        _manager->check();
    }
}

void
SubscriberPoolMonitor::startMonitor()
{
    Lock sync(*this);
    if(!_needCheck)
    {
        _needCheck = true;
        notify();
    }
}

void
SubscriberPoolMonitor::stopMonitor()
{
    Lock sync(*this);
    _needCheck = false;
}

void
SubscriberPoolMonitor::destroy()
{
    Lock sync(*this);
    _destroyed = true;
    notify();
}

SubscriberPool::SubscriberPool(const InstancePtr& instance) :
    _traceLevels(instance->traceLevels()),
    _sizeMax(instance->properties()->getPropertyAsIntWithDefault("IceStorm.SubscriberPool.SizeMax", 0)),
    _sizeWarn(instance->properties()->getPropertyAsIntWithDefault("IceStorm.SubscriberPool.SizeWarn", 0)),
    _size(instance->properties()->getPropertyAsIntWithDefault("IceStorm.SubscriberPool.Size", 1)),
    // minimum 50ms, default 1s.
    _timeout(IceUtil::Time::milliSeconds(max(instance->properties()->getPropertyAsIntWithDefault(
						 "IceStorm.SubscriberPool.Timeout", 1000), 50))),
    // 10 * the stall timeout.
    _stallCheck(_timeout * 10),
    _destroyed(false),
    _reap(0)
{
    try
    {
	__setNoDelete(true);
	_subscriberPoolMonitor = new SubscriberPoolMonitor(this, _timeout);
	for(unsigned int i = 0; i < _size; ++i)
	{
	    ++_inUse;
	    _workers.push_back(new SubscriberPoolWorker(this));
	}
    }
    catch(const IceUtil::Exception& ex)
    {
	{
	    Ice::Error out(_traceLevels->logger);
	    out << "SubscriberPool: " << ex;
	}
	destroy();
	__setNoDelete(false);
	throw;
    }

    __setNoDelete(false);
}

SubscriberPool::~SubscriberPool()
{
}
    
void
SubscriberPool::flush(list<SubscriberPtr>& subscribers)
{
    Lock sync(*this);
    if(_destroyed)
    {
	return;
    }
    //
    // Splice on the new set of subscribers to SubscriberPool.
    //
    _pending.splice(_pending.end(), subscribers);
    assert(invariants());
    notifyAll();
}

void
SubscriberPool::flush(const SubscriberPtr& subscriber)
{
    Lock sync(*this);
    if(_destroyed)
    {
	return;
    }
    _pending.push_back(subscriber);
    assert(invariants());
    notify();
}

void
SubscriberPool::add(const SubscriberPtr& subscriber)
{
    Lock sync(*this);
    if(_destroyed)
    {
	return;
    }
    _subscribers.push_back(subscriber);
    assert(invariants());
}

void
SubscriberPool::remove(const SubscriberPtr& subscriber)
{
    Lock sync(*this);
    if(_destroyed)
    {
	return;
    }
    //
    // Note that this cannot remove based on the subscriber id because
    // the pool is TopicManager scoped and not topic scoped therefore
    // its quite possible to have two subscribers with the same id in
    // the list.
    //
    list<SubscriberPtr>::iterator p = find(_subscribers.begin(), _subscribers.end(), subscriber);
    assert(p != _subscribers.end());
    _subscribers.erase(p);
}

//
// The passed subscriber need to be enqueued again.
//
void
SubscriberPool::dequeue(SubscriberPtr& subscriber, bool requeue, const IceUtil::Time& interval, bool& computeInterval)
{
    Lock sync(*this);

    if(_destroyed)
    {
	subscriber = 0;
	return;
    }

    if(subscriber)
    {
	if(requeue)
	{
	    _pending.push_back(subscriber);
	    assert(invariants());
	}
	subscriber->flushTime(interval);
    }
    //
    // Clear the reference.
    //
    subscriber = 0;

    //
    // The worker is no longer in use.
    //
    --_inUse;

    //
    // If _sizeMax is 1 we never spawn up new threads if a stall is
    // detected.
    //
    if(_sizeMax != 1)
    {

	//
	// Reap dead workers, if necessary.
	//
	if(_reap > 0)
	{
	    if(_traceLevels->subscriberPool > 0)
	    {
		Ice::Trace out(_traceLevels->logger, _traceLevels->subscriberPoolCat);
		out << "reaping: " << _reap << " workers";
	    }
	    list<IceUtil::ThreadPtr>::iterator p = _workers.begin();
	    while(p != _workers.end() && _reap > 0)
	    {
		if(!(*p)->isAlive())
		{
		    (*p)->getThreadControl().join();
		    p = _workers.erase(p);
		    --_reap;
		}
		else
		{
		    ++p;
		}
	    }
	}

	//
	// If we have extra workers every _stallCheck period we run
	// through the complete set of subscribers and determine how
	// many have stalled since the last check. If this number is
	// less than the number of extra threads then we terminate the
	// calling worker.
	//
	// - The flush time is protected by the subscriber pool mutex.
	// - The flush time is only computed if we have extra threads,
	// otherwise it is set to some large value.
	// - The max flush time is reset to the next sending interval
	// after after _stallCheck period.
	// - Every subscriber is considered to be stalled iff it has
	// never sent an event or we have just created the first
	// additional worker. The first handles the case where a
	// subscriber stalls for a long time on the first message
	// send. The second means that we can disable computation of
	// the flush latency if there are no additional threads.
	//
	if(_workers.size() > _size)
	{
	    IceUtil::Time now = IceUtil::Time::now();
	    if(now - _lastStallCheck > _stallCheck)
	    {
		_lastStallCheck = now;
		unsigned int stalls = 0;
		for(list<SubscriberPtr>::const_iterator p = _subscribers.begin(); p != _subscribers.end(); ++p)
		{
		    if((*p)->pollMaxFlushTime(now) > _timeout)
		    {
			++stalls;
		    }
		}

		if(_traceLevels->subscriberPool > 0)
		{
		    Ice::Trace out(_traceLevels->logger, _traceLevels->subscriberPoolCat);
		    out << "checking stalls. extra workers: " << _workers.size() - _size
			<< " subscribers: " << _subscribers.size() << " stalls: " << stalls;
		}

		if((_workers.size() - _size) > stalls)
		{
		    if(_traceLevels->subscriberPool > 0)
		    {
			Ice::Trace out(_traceLevels->logger, _traceLevels->subscriberPoolCat);
			out << "destroying workers";
		    }
		    ++_reap;
		    return;
		}
	    }
	}
    }
	
    while(_pending.empty() && !_destroyed)
    {
	//
	// If we wait then there is no need to monitor anymore.
	//
	_subscriberPoolMonitor->stopMonitor();
	wait();
    }

    if(_destroyed)
    {
	return;
    }

    _lastDequeue = IceUtil::Time::now();

    subscriber = _pending.front();
    _pending.pop_front();

    ++_inUse;

    //
    // If all threads are now in use then we need to start the
    // monitoring, otherwise we don't need to monitor.
    //
    if(_inUse == _workers.size() && (_workers.size() < _sizeMax || _sizeMax != 1))
    {
	_subscriberPoolMonitor->startMonitor();
    }
    else
    {
	_subscriberPoolMonitor->stopMonitor();
    }
    //
    // We only need to compute the push interval if we've created
    // stall threads.
    //
    computeInterval = (_workers.size() - _size) > 0;
}

void
SubscriberPool::destroy()
{
    //
    // First mark the pool as destroyed. This causes all of the worker
    // threads to unblock and terminate. We also clear the set of
    // subscribers here since there is a cycle (instance -> pool ->
    // subscribers -> instance).  No new subscribers can be added once
    // _destroyed is set.
    //
    {
	Lock sync(*this);
	_destroyed = true;
	notifyAll();
	if(_subscriberPoolMonitor)
	{
	    _subscriberPoolMonitor->destroy();
	}
	_subscribers.clear();
	_pending.clear();
    }
    //
    // Next join with each worker.
    //
    for(list<IceUtil::ThreadPtr>::const_iterator p = _workers.begin(); p != _workers.end(); ++p)
    {
	(*p)->getThreadControl().join();
    }
    _workers.clear();
    
    //
    // Once all of the workers have gone then we'll no longer have
    // concurrent access to the pool monitor, so we can join with it
    // and then clear to remove the circular reference count.
    //
    if(_subscriberPoolMonitor)
    {
	_subscriberPoolMonitor->getThreadControl().join();
	_subscriberPoolMonitor = 0;
    }
}

void
SubscriberPool::check()
{
    Lock sync(*this);
    if(_destroyed)
    {
	return;
    }

    IceUtil::Time now = IceUtil::Time::now();
    IceUtil::Time interval = now - _lastDequeue;
/*
    if(_traceLevels->subscriberPool > 1)
    {
	Ice::Trace out(_traceLevels->logger, _traceLevels->subscriberPoolCat);
	out << "check called: interval: "  << interval << " timeout: " << _timeout
	    << " pending: " << _pending.size() << " running: " << _workers.size()
	    << " sizeMax: " << _sizeMax;
    }
*/

    if(interval > _timeout && _pending.size() > 0 && (_workers.size() < _sizeMax || _sizeMax == 0))
    {
	if(_traceLevels->subscriberPool > 0)
	{
	    Ice::Trace out(_traceLevels->logger, _traceLevels->subscriberPoolCat);
	    out << "detected stall: creating thread: threads: " << _workers.size();
	}
	
	//
	// We'll now start stall checking at regular intervals if this
	// is the first newly created worker. Here we need to
	// initially set the stall check and the number of requests at
	// this point.
	//
	if(_workers.size() == _size)
	{
	    _lastStallCheck = now;
	}

	++_inUse;
	_workers.push_back(new SubscriberPoolWorker(this));
    }
}

bool
SubscriberPool::invariants()
{
    set<SubscriberPtr> subs;
    list<SubscriberPtr>::const_iterator p;
    for(p = _subscribers.begin(); p != _subscribers.end(); ++p)
    {
	assert(subs.find(*p) == subs.end());
	subs.insert(*p);
    }
    subs.clear();
    for(p = _pending.begin(); p != _pending.end(); ++p)
    {
	assert(subs.find(*p) == subs.end());
	subs.insert(*p);
    }
    return true;
}