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.

Static Private Member Functions
static int	getJobLimit (JobType type)

Private Attributes
beast::Journal	m_journal

std::mutex	m_mutex

std::uint64_t	m_lastJob {0}

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 {0}

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 37 of file JobQueue.h.

Member Typedef Documentation

◆ JobFunction

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

Definition at line 123 of file JobQueue.h.

◆ JobDataMap

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

private

Definition at line 223 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 44 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 147 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 390 of file JobQueue.h.

◆ getJobCount()

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

Jobs waiting at this priority.

Definition at line 105 of file JobQueue.cpp.

◆ getJobCountTotal()

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

Jobs waiting plus running at this priority.

Definition at line 115 of file JobQueue.cpp.

◆ getJobCountGE()

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

All waiting jobs at or greater than this priority.

Definition at line 125 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 142 of file JobQueue.cpp.

◆ addLoadEvents()

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

Add multiple load events.

Definition at line 154 of file JobQueue.cpp.

◆ isOverloaded()

bool xrpl::JobQueue::isOverloaded ( )

Definition at line 165 of file JobQueue.cpp.

◆ getJson()

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

Definition at line 173 of file JobQueue.cpp.

◆ rendezvous()

void xrpl::JobQueue::rendezvous ( )

Block until no jobs running.

Definition at line 230 of file JobQueue.cpp.

◆ stop()

void xrpl::JobQueue::stop ( )

Definition at line 251 of file JobQueue.cpp.

◆ isStopping()

bool xrpl::JobQueue::isStopping ( ) const

Definition at line 210 of file JobQueue.h.

◆ isStopped()

bool xrpl::JobQueue::isStopped ( ) const

Definition at line 272 of file JobQueue.cpp.

◆ collect()

void xrpl::JobQueue::collect ( )

private

Definition at line 51 of file JobQueue.cpp.

◆ getJobTypeData()

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

private

Definition at line 237 of file JobQueue.cpp.

◆ addRefCountedJob()

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

private

Definition at line 58 of file JobQueue.cpp.

◆ getNextJob()

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

private

Definition at line 278 of file JobQueue.cpp.

◆ finishJob()

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

private

Definition at line 308 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 329 of file JobQueue.cpp.

◆ getJobLimit()

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

staticprivate

Definition at line 380 of file JobQueue.cpp.

Friends And Related Symbol Documentation

◆ Coro

friend class Coro

friend

Definition at line 221 of file JobQueue.h.

Member Data Documentation

◆ m_journal

beast::Journal xrpl::JobQueue::m_journal

private

Definition at line 225 of file JobQueue.h.

◆ m_mutex

std::mutex xrpl::JobQueue::m_mutex

mutableprivate

Definition at line 226 of file JobQueue.h.

◆ m_lastJob

std::uint64_t xrpl::JobQueue::m_lastJob {0}

private

Definition at line 227 of file JobQueue.h.

◆ m_jobSet

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

private

Definition at line 228 of file JobQueue.h.

◆ jobCounter_

JobCounter xrpl::JobQueue::jobCounter_

private

Definition at line 229 of file JobQueue.h.

◆ stopping_

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

private

Definition at line 230 of file JobQueue.h.

◆ stopped_

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

private

Definition at line 231 of file JobQueue.h.

◆ m_jobData

JobDataMap xrpl::JobQueue::m_jobData

private

Definition at line 232 of file JobQueue.h.

◆ m_invalidJobData

JobTypeData xrpl::JobQueue::m_invalidJobData

private

Definition at line 233 of file JobQueue.h.

◆ m_processCount

int xrpl::JobQueue::m_processCount {0}

private

Definition at line 236 of file JobQueue.h.

◆ nSuspend_

int xrpl::JobQueue::nSuspend_ = 0

private

Definition at line 239 of file JobQueue.h.

◆ m_workers

Workers xrpl::JobQueue::m_workers

private

Definition at line 241 of file JobQueue.h.

◆ perfLog_

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

private

Definition at line 244 of file JobQueue.h.

◆ m_collector

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

private

Definition at line 245 of file JobQueue.h.

◆ job_count

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

private

Definition at line 246 of file JobQueue.h.

◆ hook

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

private

Definition at line 247 of file JobQueue.h.

◆ cv_

std::condition_variable xrpl::JobQueue::cv_

private

Definition at line 249 of file JobQueue.h.

Classes

Public Types

Public Member Functions

Private Types

Private Member Functions

Static 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_