A pool of threads to perform work. More...

#include <JobQueue.h>

Inheritance diagram for xrpl::JobQueue:

Collaboration diagram for xrpl::JobQueue:

[legend]

Classes
class	Coro
	Coroutines must run to completion. More...

Public Types
using	JobFunction = std::function< void()>

Public Member Functions
	JobQueue (int threadCount, beast::insight::Collector::ptr const &collector, beast::Journal journal, Logs &logs, perf::PerfLog &perfLog)

	~JobQueue ()

template<typename JobHandler >
bool	addJob (JobType type, std::string const &name, JobHandler &&jobHandler)
	Adds a job to the JobQueue.

template<class F >
std::shared_ptr< Coro >	postCoro (JobType t, std::string const &name, F &&f)
	Creates a coroutine and adds a job to the queue which will run it.

int	getJobCount (JobType t) const
	Jobs waiting at this priority.

int	getJobCountTotal (JobType t) const
	Jobs waiting plus running at this priority.

int	getJobCountGE (JobType t) const
	All waiting jobs at or greater than this priority.

std::unique_ptr< LoadEvent >	makeLoadEvent (JobType t, std::string const &name)
	Return a scoped LoadEvent.

void	addLoadEvents (JobType t, int count, std::chrono::milliseconds elapsed)
	Add multiple load events.

bool	isOverloaded ()

Json::Value	getJson (int c=0)

void	rendezvous ()
	Block until no jobs running.

void	stop ()

bool	isStopping () const

bool	isStopped () const

Private Types
using	JobDataMap = std::map< JobType, JobTypeData >

Private Member Functions
void	collect ()

JobTypeData &	getJobTypeData (JobType type)

bool	addRefCountedJob (JobType type, std::string const &name, JobFunction const &func)

void	getNextJob (Job &job)

void	finishJob (JobType type)

void	processTask (int instance) override
	Perform a task.

int	getJobLimit (JobType type)

Private Attributes
beast::Journal	m_journal

std::mutex	m_mutex

std::uint64_t	m_lastJob

std::set< Job >	m_jobSet

JobCounter	jobCounter_

std::atomic_bool	stopping_ {false}

std::atomic_bool	stopped_ {false}

JobDataMap	m_jobData

JobTypeData	m_invalidJobData

int	m_processCount

int	nSuspend_ = 0

Workers	m_workers

perf::PerfLog &	perfLog_

beast::insight::Collector::ptr	m_collector

beast::insight::Gauge	job_count

beast::insight::Hook	hook

std::condition_variable	cv_

Friends
class	Coro

Detailed Description

A pool of threads to perform work.

A job posted will always run to completion.

Coroutines that are suspended must be resumed, and run to completion.

When the JobQueue stops, it waits for all jobs and coroutines to finish.

Definition at line 36 of file JobQueue.h.

Member Typedef Documentation

◆ JobFunction

using xrpl::JobQueue::JobFunction = std::function<void()>

Definition at line 122 of file JobQueue.h.

◆ JobDataMap

using xrpl::JobQueue::JobDataMap = std::map<JobType, JobTypeData>

private

Definition at line 221 of file JobQueue.h.

Constructor & Destructor Documentation

◆ JobQueue()

xrpl::JobQueue::JobQueue	(	int	threadCount,
		beast::insight::Collector::ptr const &	collector,
		beast::Journal	journal,
		Logs &	logs,
		perf::PerfLog &	perfLog
	)

Definition at line 9 of file JobQueue.cpp.

◆ ~JobQueue()

xrpl::JobQueue::~JobQueue ( )

Definition at line 46 of file JobQueue.cpp.

Member Function Documentation

◆ addJob()

template<typename JobHandler >

bool xrpl::JobQueue::addJob	(	JobType	type,
		std::string const &	name,
		JobHandler &&	jobHandler
	)

Adds a job to the JobQueue.

Parameters

type	The type of job.
name	Name of the job.
jobHandler	Lambda with signature void (Job&). Called when the job is executed.

Returns: true if jobHandler added to queue.

Definition at line 145 of file JobQueue.h.

◆ postCoro()

template<class F >

std::shared_ptr< JobQueue::Coro > xrpl::JobQueue::postCoro	(	JobType	t,
		std::string const &	name,
		F &&	f
	)

Creates a coroutine and adds a job to the queue which will run it.

Parameters

t	The type of job.
name	Name of the job.
f	Has a signature of void(std::shared_ptr<Coro>). Called when the job executes.

Returns: shared_ptr to posted Coro. nullptr if post was not successful.

Definition at line 386 of file JobQueue.h.

◆ getJobCount()

int xrpl::JobQueue::getJobCount ( JobType t ) const

Jobs waiting at this priority.

Definition at line 106 of file JobQueue.cpp.

◆ getJobCountTotal()

int xrpl::JobQueue::getJobCountTotal ( JobType t ) const

Jobs waiting plus running at this priority.

Definition at line 116 of file JobQueue.cpp.

◆ getJobCountGE()

int xrpl::JobQueue::getJobCountGE ( JobType t ) const

All waiting jobs at or greater than this priority.

Definition at line 126 of file JobQueue.cpp.

◆ makeLoadEvent()

std::unique_ptr< LoadEvent > xrpl::JobQueue::makeLoadEvent	(	JobType	t,
		std::string const &	name
	)

Return a scoped LoadEvent.

Definition at line 143 of file JobQueue.cpp.

◆ addLoadEvents()

void xrpl::JobQueue::addLoadEvents	(	JobType	t,
		int	count,
		std::chrono::milliseconds	elapsed
	)

Add multiple load events.

Definition at line 155 of file JobQueue.cpp.

◆ isOverloaded()

bool xrpl::JobQueue::isOverloaded ( )

Definition at line 166 of file JobQueue.cpp.

◆ getJson()

Json::Value xrpl::JobQueue::getJson ( int c = 0 )

Definition at line 172 of file JobQueue.cpp.

◆ rendezvous()

void xrpl::JobQueue::rendezvous ( )

Block until no jobs running.

Definition at line 229 of file JobQueue.cpp.

◆ stop()

void xrpl::JobQueue::stop ( )

Definition at line 250 of file JobQueue.cpp.

◆ isStopping()

bool xrpl::JobQueue::isStopping ( ) const

Definition at line 208 of file JobQueue.h.

◆ isStopped()

bool xrpl::JobQueue::isStopped ( ) const

Definition at line 271 of file JobQueue.cpp.

◆ collect()

void xrpl::JobQueue::collect ( )

private

Definition at line 53 of file JobQueue.cpp.

◆ getJobTypeData()

JobTypeData & xrpl::JobQueue::getJobTypeData ( JobType type )

private

Definition at line 236 of file JobQueue.cpp.

◆ addRefCountedJob()

bool xrpl::JobQueue::addRefCountedJob	(	JobType	type,
		std::string const &	name,
		JobFunction const &	func
	)

private

Definition at line 60 of file JobQueue.cpp.

◆ getNextJob()

void xrpl::JobQueue::getNextJob ( Job & job )

private

Definition at line 277 of file JobQueue.cpp.

◆ finishJob()

void xrpl::JobQueue::finishJob ( JobType type )

private

Definition at line 306 of file JobQueue.cpp.

◆ processTask()

void xrpl::JobQueue::processTask ( int instance )

overrideprivatevirtual

Perform a task.

The call is made on a thread owned by Workers. It is important that you only process one task from inside your callback. Each call to addTask will result in exactly one call to processTask.

Parameters

instance The worker thread instance.

See also: Workers::addTask

Implements xrpl::Workers::Callback.

Definition at line 325 of file JobQueue.cpp.

◆ getJobLimit()

int xrpl::JobQueue::getJobLimit ( JobType type )

private

Definition at line 376 of file JobQueue.cpp.

Friends And Related Symbol Documentation

◆ Coro

friend class Coro

friend

Definition at line 219 of file JobQueue.h.

Member Data Documentation

◆ m_journal

beast::Journal xrpl::JobQueue::m_journal

private

Definition at line 223 of file JobQueue.h.

◆ m_mutex

std::mutex xrpl::JobQueue::m_mutex

mutableprivate

Definition at line 224 of file JobQueue.h.

◆ m_lastJob

std::uint64_t xrpl::JobQueue::m_lastJob

private

Definition at line 225 of file JobQueue.h.

◆ m_jobSet

std::set<Job> xrpl::JobQueue::m_jobSet

private

Definition at line 226 of file JobQueue.h.

◆ jobCounter_

JobCounter xrpl::JobQueue::jobCounter_

private

Definition at line 227 of file JobQueue.h.

◆ stopping_

std::atomic_bool xrpl::JobQueue::stopping_ {false}

private

Definition at line 228 of file JobQueue.h.

◆ stopped_

std::atomic_bool xrpl::JobQueue::stopped_ {false}

private

Definition at line 229 of file JobQueue.h.

◆ m_jobData

JobDataMap xrpl::JobQueue::m_jobData

private

Definition at line 230 of file JobQueue.h.

◆ m_invalidJobData

JobTypeData xrpl::JobQueue::m_invalidJobData

private

Definition at line 231 of file JobQueue.h.

◆ m_processCount

int xrpl::JobQueue::m_processCount

private

Definition at line 234 of file JobQueue.h.

◆ nSuspend_

int xrpl::JobQueue::nSuspend_ = 0

private

Definition at line 237 of file JobQueue.h.

◆ m_workers

Workers xrpl::JobQueue::m_workers

private

Definition at line 239 of file JobQueue.h.

◆ perfLog_

perf::PerfLog& xrpl::JobQueue::perfLog_

private

Definition at line 242 of file JobQueue.h.

◆ m_collector

beast::insight::Collector::ptr xrpl::JobQueue::m_collector

private

Definition at line 243 of file JobQueue.h.

◆ job_count

beast::insight::Gauge xrpl::JobQueue::job_count

private

Definition at line 244 of file JobQueue.h.

◆ hook

beast::insight::Hook xrpl::JobQueue::hook

private

Definition at line 245 of file JobQueue.h.

◆ cv_

std::condition_variable xrpl::JobQueue::cv_

private

Definition at line 247 of file JobQueue.h.

Classes

Public Types

Public Member Functions

Private Types

Private Member Functions

Private Attributes

Friends

Detailed Description

Member Typedef Documentation

◆ JobFunction

◆ JobDataMap

Constructor & Destructor Documentation

◆ JobQueue()

◆ ~JobQueue()

Member Function Documentation

◆ addJob()

◆ postCoro()

◆ getJobCount()

◆ getJobCountTotal()

◆ getJobCountGE()

◆ makeLoadEvent()

◆ addLoadEvents()

◆ isOverloaded()

◆ getJson()

◆ rendezvous()

◆ stop()

◆ isStopping()

◆ isStopped()

◆ collect()

◆ getJobTypeData()

◆ addRefCountedJob()

◆ getNextJob()

◆ finishJob()

◆ processTask()

◆ getJobLimit()

Friends And Related Symbol Documentation

◆ Coro

Member Data Documentation

◆ m_journal

◆ m_mutex

◆ m_lastJob

◆ m_jobSet

◆ jobCounter_

◆ stopping_

◆ stopped_

◆ m_jobData

◆ m_invalidJobData

◆ m_processCount

◆ nSuspend_

◆ m_workers

◆ perfLog_

◆ m_collector

◆ job_count

◆ hook

◆ cv_