ossimJobQueue Class Reference

This is the base implementation for the job queue. More...

#include <ossimJobQueue.h>

Inheritance diagram for ossimJobQueue:

Classes
class	Callback
	The callback allows one to attach and listen for certain things. More...

Public Member Functions
	ossimJobQueue ()
	Default constructor. More...
std::shared_ptr< ossimJobQueue >	getSharedFromThis ()
	This is the safe way to create a std::shared_ptr for 'this'. More...
std::shared_ptr< const ossimJobQueue >	getSharedFromThis () const
	This is the safe way to create a std::shared_ptr for 'this'. More...
virtual void	add (std::shared_ptr< ossimJob > job, bool guaranteeUniqueFlag=true)
	Will add a job to the queue and if the guaranteeUniqueFlag is set it will scan and make sure the job is not on the queue before adding. More...
virtual std::shared_ptr< ossimJob >	removeByName (const ossimString &name)
	Allows one to remove a job passing in it's name. More...
virtual std::shared_ptr< ossimJob >	removeById (const ossimString &id)
	Allows one to remove a job passing in it's id. More...
virtual void	remove (const std::shared_ptr< ossimJob > Job)
	Allows one to pass in a job pointer to remove. More...
virtual void	removeStoppedJobs ()
	Will remove any stopped jobs from the queue. More...
virtual void	clear ()
	Will clear the queue. More...
virtual std::shared_ptr< ossimJob >	nextJob (bool blockIfEmptyFlag=true)
	Will grab the next job on the list and will return the job or a null shared_ptr. More...
virtual void	releaseBlock ()
	will release the block and have any blocked threads continue More...
bool	isEmpty () const
ossim_uint32	size ()
void	setCallback (std::shared_ptr< Callback > c)
	Allows one to set the callback to the list. More...
std::shared_ptr< Callback >	callback ()

Protected Member Functions
ossimJob::List::iterator	findById (const ossimString &id)
	Internal method that returns an iterator. More...
ossimJob::List::iterator	findByName (const ossimString &name)
	Internal method that returns an iterator. More...
ossimJob::List::iterator	findByPointer (const std::shared_ptr< ossimJob > job)
	Internal method that returns an iterator. More...
ossimJob::List::iterator	findByNameOrPointer (const std::shared_ptr< ossimJob > job)
	Internal method that returns an iterator. More...
bool	hasJob (std::shared_ptr< ossimJob > job)
	Internal method that determines if we have the job. More...

Protected Attributes
std::mutex	m_jobQueueMutex
ossim::Block	m_block
ossimJob::List	m_jobQueue
std::shared_ptr< Callback >	m_callback

Detailed Description

This is the base implementation for the job queue.

It allows one to add and remove jobs and to call the nextJob and, if specified, block the call if no jobs are on the queue. we derived from std::enable_shared_from_this which allows us access to a safe shared 'this' pointer. This is used internal to our callbacks.

The job queue is thread safe and can be shared by multiple threads.

Here is a quick code example on how to create a shared queue and to attach a thread to it. In this example we do not block the calling thread for nextJob

#include <ossim/base/Thread.h>
#include <ossim/parallel/ossimJob.h>
#include <ossim/parallel/ossimJobQueue.h>
#include <iostream>
class TestJobQueueThread : public ossim::Thread
{
public:
   TestJobQueueThread(std::shared_ptr<ossimJobQueue> q):m_q(q){}
   void run()
   {
      if(m_q)
      {
         while(true)
         {
            interrupt();
            std::shared_ptr<ossimJob> job = m_q->nextJob(false);
            if(job)
            {
               job->start();
            }
            yieldCurrentThread();
            sleepInMilliSeconds(20);
         }
      }
   }

private:
  std::shared_ptr<ossimJobQueue> m_q;
};

class TestJob : public ossimJob
{
public:
   virtual void run()
   {
      std:cout << "Running Job\n";
      ossim::Thread::sleepInSeconds(2);
      std::cout << "Finished Running Job\n";
   }
};
int main(int argc, char *argv[])
{
   std::shared_ptr<ossimJobQueue> q = std::make_shared<ossimJobQueue>();
   std::shared_ptr<TestJobQueueThread> jobQueueThread = std::make_shared<TestJobQueueThread>(q);

   jobQueueThread->start();

   q->add(std::make_shared<TestJob>());
   q->add(std::make_shared<TestJob>());
   q->add(std::make_shared<TestJob>());

   ossim::Thread::sleepInSeconds(10);
   jobQueueThread->cancel();
   jobQueueThread->waitForCompletion();
}

Definition at line 84 of file ossimJobQueue.h.

Constructor & Destructor Documentation

ossimJobQueue()

ossimJobQueue::ossimJobQueue

(

)

Default constructor.

Definition at line 5 of file ossimJobQueue.cpp.

{
}

Member Function Documentation

add()

void ossimJobQueue::add ( std::shared_ptr< ossimJob >  job, bool  guaranteeUniqueFlag = true  )

virtual

Will add a job to the queue and if the guaranteeUniqueFlag is set it will scan and make sure the job is not on the queue before adding.

Parameters

job	The job to add to the queue.
guaranteeUniqueFlag	if set to true will force a find to make sure the job does not already exist

Definition at line 9 of file ossimJobQueue.cpp.

References findByPointer(), getSharedFromThis(), m_block, m_callback, m_jobQueue, m_jobQueueMutex, and ossim::Block::set().

{
   std::shared_ptr<Callback> cb;
   {
      {
         std::lock_guard<std::mutex> lock(m_jobQueueMutex);
         
         if(guaranteeUniqueFlag)
         {
            if(findByPointer(job) != m_jobQueue.end())
            {
               m_block.set(true);
               return;
            }
         }
         cb = m_callback;
      }
      if(cb) cb->adding(getSharedFromThis(), job);
      
      job->ready();
      m_jobQueueMutex.lock();
      m_jobQueue.push_back(job);
      m_jobQueueMutex.unlock();
   }
   if(cb)
   {
      cb->added(getSharedFromThis(), job);
   }
   m_block.set(true);
}

callback()

std::shared_ptr< ossimJobQueue::Callback > ossimJobQueue::callback

(

)

Returns

the callback

Definition at line 300 of file ossimJobQueue.cpp.

References m_callback, and m_jobQueueMutex.

{
   std::lock_guard<std::mutex> lock(m_jobQueueMutex);
   return m_callback;
}

clear()

void ossimJobQueue::clear ( )

virtual

Will clear the queue.

Definition at line 143 of file ossimJobQueue.cpp.

References getSharedFromThis(), m_callback, m_jobQueue, and m_jobQueueMutex.

{
   ossimJob::List removedJobs(m_jobQueue);
   std::shared_ptr<Callback> cb;
   {
      std::lock_guard<std::mutex> lock(m_jobQueueMutex);
      m_jobQueue.clear();
      cb = m_callback;
   }
   if(cb)
   {
      for(ossimJob::List::iterator iter=removedJobs.begin();iter!=removedJobs.end();++iter)
      {
         cb->removed(getSharedFromThis(), (*iter));
      }
   }
}

findById()

ossimJob::List::iterator ossimJobQueue::findById ( const ossimString &  id )

protected

Internal method that returns an iterator.

Parameters

the	id of the job to search for

Returns

the iterator

Definition at line 216 of file ossimJobQueue.cpp.

References m_jobQueue.

Referenced by removeById().

{
   if(id.empty()) return m_jobQueue.end();
   ossimJob::List::iterator iter = m_jobQueue.begin();
   while(iter != m_jobQueue.end())
   {
      if(id == (*iter)->id())
      {
         return iter;
      }
      ++iter;
   }  
   return m_jobQueue.end();
}

findByName()

ossimJob::List::iterator ossimJobQueue::findByName ( const ossimString &  name )

protected

Internal method that returns an iterator.

Parameters

name	the name of the job to search for

Returns

the iterator

Definition at line 231 of file ossimJobQueue.cpp.

References ossimString::empty(), and m_jobQueue.

Referenced by removeByName().

{
   if(name.empty()) return m_jobQueue.end();
   ossimJob::List::iterator iter = m_jobQueue.begin();
   while(iter != m_jobQueue.end())
   {
      if(name == (*iter)->name())
      {
         return iter;
      }
      ++iter;
   }  
   return m_jobQueue.end();
}

findByNameOrPointer()

ossimJob::List::iterator ossimJobQueue::findByNameOrPointer ( const std::shared_ptr< ossimJob >  job )

protected

Internal method that returns an iterator.

Parameters

job	it will find by the name or by the pointer

Returns

the iterator

Definition at line 253 of file ossimJobQueue.cpp.

References m_jobQueue, and n.

{
   ossimString n = job->name();
   ossimJob::List::iterator iter = std::find_if(m_jobQueue.begin(), m_jobQueue.end(), [n, job](const std::shared_ptr<ossimJob> jobIter){
      bool result = (jobIter == job);
      if(!result&&!n.empty()) result = jobIter->name() == n;
      return result;
   });
   // ossimJob::List::iterator iter = m_jobQueue.begin();
   // while(iter != m_jobQueue.end())
   // {
   //    if((*iter) == job)
   //    {
   //       return iter;
   //    }
   //    else if((!n.empty())&&
   //            (job->name() == (*iter)->name()))
   //    {
   //       return iter;
   //    }
   //    ++iter;
//   }  
   
   return iter;
}

findByPointer()

ossimJob::List::iterator ossimJobQueue::findByPointer ( const std::shared_ptr< ossimJob >  job )

protected

Internal method that returns an iterator.

Parameters

job	the job to search for

Returns

the iterator

Definition at line 246 of file ossimJobQueue.cpp.

References m_jobQueue.

Referenced by add().

{
   return std::find(m_jobQueue.begin(),
                    m_jobQueue.end(),
                    job);
}

getSharedFromThis() [1/2]

std::shared_ptr<ossimJobQueue> ossimJobQueue::getSharedFromThis ( )

inline

This is the safe way to create a std::shared_ptr for 'this'.

Calls the derived method shared_from_this

Definition at line 133 of file ossimJobQueue.h.

Referenced by add(), clear(), remove(), removeById(), removeByName(), and removeStoppedJobs().

                                                   {
      return shared_from_this();
   }

getSharedFromThis() [2/2]

std::shared_ptr<const ossimJobQueue> ossimJobQueue::getSharedFromThis ( ) const

inline

This is the safe way to create a std::shared_ptr for 'this'.

Calls the derived method shared_from_this

Definition at line 141 of file ossimJobQueue.h.

                                                              {
      return shared_from_this();
   }

hasJob()

bool ossimJobQueue::hasJob ( std::shared_ptr< ossimJob >  job )

protected

Internal method that determines if we have the job.

Parameters

job	the job you wish to search for

Definition at line 279 of file ossimJobQueue.cpp.

References m_jobQueue.

{
   ossimJob::List::const_iterator iter = m_jobQueue.begin();
   while(iter != m_jobQueue.end())
   {
      if(job == (*iter))
      {
         return true;
      }
      ++iter;
   }
   
   return false;
}

isEmpty()

bool ossimJobQueue::isEmpty

(

)

const

Returns

true if the queue is empty false otherwise

Definition at line 200 of file ossimJobQueue.cpp.

References m_jobQueue, and m_jobQueueMutex.

{
   // std::lock_guard<std::mutex> lock(m_jobQueueMutex);
   // return m_jobQueue.empty();
   m_jobQueueMutex.lock();
   bool result =  m_jobQueue.empty();
   m_jobQueueMutex.unlock();
   return result;
}

nextJob()

std::shared_ptr< ossimJob > ossimJobQueue::nextJob ( bool  blockIfEmptyFlag = true )

virtual

Will grab the next job on the list and will return the job or a null shared_ptr.

You can block the caller if the queueis empty forcing it to wait for more jobs to come onto the queue

Parameters

blockIfEmptyFlag

If true it will block the calling thread until more jobs appear on the queue. If false, it will return without blocking

Returns

a shared pointer to a job

Definition at line 161 of file ossimJobQueue.cpp.

References ossim::Block::block(), m_block, m_jobQueue, m_jobQueueMutex, and ossim::Block::set().

{
   m_jobQueueMutex.lock();
   bool emptyFlag = m_jobQueue.empty();
   m_jobQueueMutex.unlock();
   if (blockIfEmptyFlag && emptyFlag)
   {
      m_block.block();
   }
   
   std::shared_ptr<ossimJob> result;
   std::lock_guard<std::mutex> lock(m_jobQueueMutex);
   
   if (m_jobQueue.empty())
   {
      m_block.set(false);
      return result;
   }
   
   ossimJob::List::iterator iter= m_jobQueue.begin();
   while((iter != m_jobQueue.end())&&
         (((*iter)->isCanceled())))
   {
      (*iter)->finished(); // mark the ob as being finished 
      iter = m_jobQueue.erase(iter);
   }
   if(iter != m_jobQueue.end())
   {
      result = *iter;
      m_jobQueue.erase(iter);
   }
   m_block.set(!m_jobQueue.empty());

   return result;
}

releaseBlock()

void ossimJobQueue::releaseBlock ( )

virtual

will release the block and have any blocked threads continue

Definition at line 196 of file ossimJobQueue.cpp.

References m_block, and ossim::Block::release().

{
   m_block.release();
}

remove()

void ossimJobQueue::remove ( const std::shared_ptr< ossimJob >  Job )

virtual

Allows one to pass in a job pointer to remove.

Parameters

job	the job you wish to remove from the list

Definition at line 87 of file ossimJobQueue.cpp.

References getSharedFromThis(), m_callback, m_jobQueue, and m_jobQueueMutex.

{
   std::shared_ptr<ossimJob> removedJob;
   std::shared_ptr<Callback> cb;
   {
      std::lock_guard<std::mutex> lock(m_jobQueueMutex);
      ossimJob::List::iterator iter = std::find(m_jobQueue.begin(), m_jobQueue.end(), job);
      if(iter!=m_jobQueue.end())
      {
         removedJob = (*iter);
         m_jobQueue.erase(iter);
      }
      cb = m_callback;
   }
   if(cb&&removedJob)
   {
      cb->removed(getSharedFromThis(), removedJob);
   }
}

removeById()

std::shared_ptr< ossimJob > ossimJobQueue::removeById ( const ossimString &  id )

virtual

Allows one to remove a job passing in it's id.

Parameters

id	The job id

Returns

a shared_ptr to the job. This will be nullptr if not found.

Definition at line 64 of file ossimJobQueue.cpp.

References findById(), getSharedFromThis(), m_block, m_callback, m_jobQueue, m_jobQueueMutex, and ossim::Block::set().

{
   std::shared_ptr<ossimJob> result;
   std::shared_ptr<Callback> cb;
   if(id.empty()) return result;
   {
      std::lock_guard<std::mutex> lock(m_jobQueueMutex);
      ossimJob::List::iterator iter = findById(id);
      if(iter!=m_jobQueue.end())
      {
         result = *iter;
         m_jobQueue.erase(iter);
      }
      cb = m_callback;
      m_block.set(!m_jobQueue.empty());
   }
   if(cb&&result)
   {
      cb->removed(getSharedFromThis(), result);
   }
   return result;
}

removeByName()

std::shared_ptr< ossimJob > ossimJobQueue::removeByName ( const ossimString &  name )

virtual

Allows one to remove a job passing in it's name.

Parameters

name	The job name

Returns

a shared_ptr to the job. This will be nullptr if not found.

Definition at line 41 of file ossimJobQueue.cpp.

References ossimString::empty(), findByName(), getSharedFromThis(), m_block, m_callback, m_jobQueue, m_jobQueueMutex, and ossim::Block::set().

{
   std::shared_ptr<ossimJob> result;
   std::shared_ptr<Callback> cb;
   if(name.empty()) return result;
   {
      std::lock_guard<std::mutex> lock(m_jobQueueMutex);
      ossimJob::List::iterator iter = findByName(name);
      if(iter!=m_jobQueue.end())
      {
         result = *iter;
         m_jobQueue.erase(iter);
      }
      cb = m_callback;
   }      
   m_block.set(!m_jobQueue.empty());
   
   if(cb&&result)
   {
      cb->removed(getSharedFromThis(), result);
   }
   return result;
}

removeStoppedJobs()

void ossimJobQueue::removeStoppedJobs ( )

virtual

Will remove any stopped jobs from the queue.

Definition at line 107 of file ossimJobQueue.cpp.

References getSharedFromThis(), m_callback, m_jobQueue, and m_jobQueueMutex.

{
   ossimJob::List removedJobs;
   std::shared_ptr<Callback> cb;
   {
      std::lock_guard<std::mutex> lock(m_jobQueueMutex);
      cb = m_callback;
      ossimJob::List::iterator iter = m_jobQueue.begin();
      while(iter!=m_jobQueue.end())
      {
         if((*iter)->isStopped())
         {
            removedJobs.push_back(*iter);
            iter = m_jobQueue.erase(iter);
         }
         else 
         {
            ++iter;
         }
      }
   }
   if(!removedJobs.empty())
   {
      if(cb)
      {
         ossimJob::List::iterator iter = removedJobs.begin();
         while(iter!=removedJobs.end())
         {
            cb->removed(getSharedFromThis(), (*iter));
            ++iter;
         }
      }
      removedJobs.clear();
   }
}

setCallback()

void ossimJobQueue::setCallback

(

std::shared_ptr< Callback >

c

)

Allows one to set the callback to the list.

Parameters

c	shared_ptr to a callback

Definition at line 294 of file ossimJobQueue.cpp.

References m_callback, and m_jobQueueMutex.

{
   std::lock_guard<std::mutex> lock(m_jobQueueMutex);
   m_callback = c;
}

size()

ossim_uint32 ossimJobQueue::size

(

)

Returns

the number of jobs on the queue

Definition at line 210 of file ossimJobQueue.cpp.

References m_jobQueue, and m_jobQueueMutex.

{
   std::lock_guard<std::mutex> lock(m_jobQueueMutex);
   return (ossim_uint32) m_jobQueue.size();
}

Member Data Documentation

m_block

ossim::Block ossimJobQueue::m_block

protected

Definition at line 267 of file ossimJobQueue.h.

Referenced by add(), nextJob(), releaseBlock(), removeById(), and removeByName().

m_callback

std::shared_ptr<Callback> ossimJobQueue::m_callback

protected

Definition at line 269 of file ossimJobQueue.h.

Referenced by add(), callback(), clear(), remove(), removeById(), removeByName(), removeStoppedJobs(), and setCallback().

m_jobQueue

ossimJob::List ossimJobQueue::m_jobQueue

protected

Definition at line 268 of file ossimJobQueue.h.

Referenced by add(), clear(), findById(), findByName(), findByNameOrPointer(), findByPointer(), hasJob(), isEmpty(), nextJob(), remove(), removeById(), removeByName(), removeStoppedJobs(), and size().

m_jobQueueMutex

std::mutex ossimJobQueue::m_jobQueueMutex

mutableprotected

Definition at line 266 of file ossimJobQueue.h.

Referenced by add(), callback(), clear(), isEmpty(), nextJob(), remove(), removeById(), removeByName(), removeStoppedJobs(), setCallback(), and size().

---

The documentation for this class was generated from the following files:

• ossimJobQueue.h
• ossimJobQueue.cpp