ReducingTaskTTLManager.hpp

Go to the documentation of this file.

#ifndef REDUCINGTASKTTLMANAGER_HPP_
#define REDUCINGTASKTTLMANAGER_HPP_

#include <map>
#include <vector>
#include <set>
#include <Poco/Timestamp.h>

namespace HCE
{
namespace reduce
{

const int ONE_THOUSAND_MICROSECONDS = 1000;
const unsigned int USE_TASK_TTL_VALUE = 0;

class ReducingTaskTTLManager {
public:    
  void addTaskTTL( unsigned long long taskId, unsigned int ttl ) {
    //if task is already terminated, don't add it
    if( isTaskTerminatedByTTL( taskId ) ) {
      return;
    }
    
    auto iter = _runningTasks.find( taskId );
    
    if( iter == _runningTasks.end() ) {
      //insert the task TTL info
      _runningTasks[ taskId ] = std::make_pair( Poco::Timestamp(), ttl );
    } else {
      //simply update the TTL value if the task is already inserted
      iter->second.second = ttl;
    }
  }
  
  void removeTaskTTL( unsigned long long taskId ) {
    auto iter = _runningTasks.find( taskId );
    if( iter != _runningTasks.end() ) {
      //if it is expired by TTL, move it to the terminated TTL queue
      if( iter->second.first.isElapsed( (iter->second.second) * ONE_THOUSAND_MICROSECONDS ) ) { 
        addTerminatedTask(taskId );
      }
      _runningTasks.erase( iter );
    }    
  }
  
  unsigned int getTaskTTL( unsigned long long taskId ) const {
    auto iter = _runningTasks.find( taskId );
    if( iter != _runningTasks.end() ) {
      return iter->second.second;
    }
    return 0;
  }
  
  
  int cleanupExpiredTasksByTTL(unsigned int maxRemoveTasks, unsigned int ttl = USE_TASK_TTL_VALUE) {
    std::vector< unsigned long long > expiredTasks;
    
    getExceededTTLTasks( maxRemoveTasks, expiredTasks, ttl );
    
    for( int i = 0, n = expiredTasks.size(); i < n; i++ ) {
      _runningTasks.erase( expiredTasks[i] );
      //insert into the terminated tasks
      addTerminatedTask( expiredTasks[i] );
    }
    
    return expiredTasks.size();
  }
  
  bool isTaskTerminatedByTTL( unsigned long long taskId ) const {
    return _terminatedTasks.find( taskId ) != _terminatedTasks.end();
  }
  
  unsigned int cleanupExpiredTasksByTTLQueue( unsigned int maxRemoveItems, unsigned int minTerminateTime ) {
    unsigned int i = 0, n = _terminatedTasksTime.size();
    
    for( ; i < n && i < maxRemoveItems && _terminatedTasksTime[i].second.isElapsed( minTerminateTime * ONE_THOUSAND_MICROSECONDS ); i++ );    
    
    if( i > 0 ) {
      for(unsigned int j = 0; j < i; j++ ) {
        _terminatedTasks.erase( _terminatedTasksTime[j].first );
      }    
      _terminatedTasksTime.erase( _terminatedTasksTime.begin(), _terminatedTasksTime.begin() + i );
    }
    return i;
  }
  
  bool isTaskExpired( unsigned long long taskId, unsigned int ttl = USE_TASK_TTL_VALUE) {
    std::map< unsigned long long, std::pair< Poco::Timestamp, unsigned int > >::const_iterator iter = _runningTasks.find( taskId );
    
    return (iter == _runningTasks.end() )? false:  iter->second.first.isElapsed( ( ( ttl > 0 )?ttl:iter->second.second ) * ONE_THOUSAND_MICROSECONDS );                
  }
  
  void getExceededTTLTasks( unsigned int maxTaskNumber, std::vector<unsigned long long>& exceededTTLTasks, unsigned int ttl = 0 ) const {
    std::map< unsigned long long, std::pair< Poco::Timestamp, unsigned int > >::const_iterator iter;
    unsigned int taskCount = 0;
    
    for( iter = _runningTasks.begin(); iter != _runningTasks.end(); ++iter ) {
      if( iter->second.first.isElapsed( (( ttl > 0 )?ttl: iter->second.second )*ONE_THOUSAND_MICROSECONDS ) ) {
        exceededTTLTasks.push_back( iter->first );
        taskCount++;
        if( taskCount >= maxTaskNumber ) {
          break;
        }
      }
    }
  }
private:  
  void addTerminatedTask( unsigned long long taskId ) {
    _terminatedTasks.insert( taskId );
    _terminatedTasksTime.push_back( std::make_pair( taskId, Poco::Timestamp() ) );
  }
private:
  //map between <taskId, <task create time, task ttl > >
  std::map< unsigned long long, std::pair< Poco::Timestamp, unsigned int > > _runningTasks;
  std::set< unsigned long long > _terminatedTasks;
  //vector of <taskid, task terminate time>
  std::vector< std::pair< unsigned long long, Poco::Timestamp > > _terminatedTasksTime;
  
};

}
}

#endif/*REDUCINGTASKTTLMANAGER_HPP_*/