worker.cpp File Reference

#include <algorithm>
#include <cassert>
#include <pthread.h>
#include <queue>
#include <set>
#include <vector>
#include "utils/macros.h"
#include "utils/misc.h"
#include "utils/thread.h"
#include "utils/worker.h"

Include dependency graph for worker.cpp:

Go to the source code of this file.

Classes
struct	respawn_parallel_data
class	worker_pool
struct	worker_state_observer
struct	wp_task_data
	Holds data to be passed to worker. More...
struct	wp_worker
	This class represents a worker that is called from within the pool. More...

Functions
static void	func_delete (wp_worker *arg)
void	respawn_parallel (void *in, void *out, size_t nmemb, size_t size, respawn_parallel_callback_t c, void *udata)
static void *	respawn_parallel_task (void *arg)
task_result_handle_t	task_run_async (runnable_t task, void *data)
	Runs task asynchronously.
void	task_run_async_detached (runnable_t task, void *data)
	Runs task asynchronously in a detached state.
void	task_run_parallel (runnable_t task, int worker_count, void *data, size_t data_size, void **res)
void *	wait_task (task_result_handle_t handle)

Variables
static class worker_pool	instance

Detailed Description

Author

Martin Pulec pulec.nosp@m.@ces.nosp@m.net.c.nosp@m.z

Definition in file worker.cpp.

Function Documentation

func_delete()

static void func_delete ( wp_worker * arg )

static

Definition at line 179 of file worker.cpp.

                                        {
        delete arg;
}

Referenced by worker_pool::~worker_pool().

respawn_parallel()

void respawn_parallel	(	void *	in,
		void *	out,
		size_t	nmemb,
		size_t	size,
		respawn_parallel_callback_t	c,
		void *	udata )

Automatically respawns threads to convert in to out

Botn input and output elements must currently have the same size (can be changed in future). Option semantics is similar to qsort().

Definition at line 332 of file worker.cpp.

{
        const int threads = min<int>(get_cpu_core_count(), MAX_CPU_CORES);
        struct respawn_parallel_data data[MAX_CPU_CORES];
 
        for (int i = 0; i < threads; ++i) {
                data[i].c = c;
                data[i].in = (char *) in + i * (nmemb / threads) * size;
                data[i].out = (char *) out + i * (nmemb / threads) * size;
                data[i].data_len = (nmemb / threads) * size;
                data[i].udata = udata;
                if (i == threads - 1) {
                        data[i].data_len = size * (nmemb - (threads - 1) * (nmemb / threads));
                }
        }
 
        task_run_parallel(respawn_parallel_task, threads, data, sizeof data[0], NULL);
}

References respawn_parallel_data::c, respawn_parallel_data::data_len, get_cpu_core_count(), respawn_parallel_data::in, MAX_CPU_CORES, min, respawn_parallel_data::out, respawn_parallel_task(), task_run_parallel(), and respawn_parallel_data::udata.

Referenced by r10k_full_to_limited(), and r10k_limited_to_full().

Here is the call graph for this function:

respawn_parallel_task()

static void * respawn_parallel_task ( void * arg )

static

Definition at line 321 of file worker.cpp.

                                              {
        auto data = (struct respawn_parallel_data *) arg;
        data->c(data->in, data->out, data->data_len, data->udata);
        return NULL;
}

Referenced by respawn_parallel().

task_run_async()

task_result_handle_t task_run_async	(	runnable_t	task,
		void *	data )

Runs task asynchronously.

Parameters

task	callback to be run
data	additional data to be passed to the callback

Returns

handle to the task

Note

If you use this call wait_task() must be run.

Definition at line 262 of file worker.cpp.

{
        return instance.run_async(task, data, false);
}

References instance, and worker_pool::run_async().

Referenced by state_capture_filter_gamma::apply_lut(), compress_frame_tiles(), decompress_thread(), task_run_parallel(), and video_reading_thread().

Here is the call graph for this function:

task_run_async_detached()

void task_run_async_detached	(	runnable_t	task,
		void *	data )

Runs task asynchronously in a detached state.

Detached task should own its resources. Moreover, it must not use any static variables/objects.

Parameters

task	callback to be run
data	additional data to be passed to the callback

Definition at line 275 of file worker.cpp.

{
        instance.run_async(task, data, true);
}

References instance, and worker_pool::run_async().

Referenced by decode_audio_frame(), process_statistics(), and ultragrid_rtp_video_rxtx::send_frame().

Here is the call graph for this function:

task_run_parallel()

void task_run_parallel	(	runnable_t	task,
		int	worker_count,
		void *	data,
		size_t	data_size,
		void **	res )

This combines task_run_async() + wait_task()

Parameters

task	task to be run
worker_count	number of workers to be run
data	pointer to data array to be passed to task
data_size	size of element of data
res	(optional) pointer to result array, may be NULL

Definition at line 294 of file worker.cpp.

{
        if (worker_count == 1) {
                task(data);
                return;
        }
 
        vector<task_result_handle_t> tasks(worker_count);
        for (int i = 0; i < worker_count; ++i) {
                tasks[i] = task_run_async(task, (void *)((char *) data + i * data_size));
        }
        for (int i = 0; i < worker_count; ++i) {
                if (res != nullptr) {
                        res[i] = wait_task(tasks[i]);
                } else {
                        wait_task(tasks[i]);
                }
        }
}

References task_run_async(), and wait_task().

Referenced by av_to_uv_convert(), parallel_pix_conv(), respawn_parallel(), and to_lavc_vid_conv().

Here is the call graph for this function:

wait_task()

void * wait_task

(

task_result_handle_t

handle

)

Definition at line 280 of file worker.cpp.

{
        return instance.wait_task(handle);
}

References instance, and worker_pool::wait_task().

Referenced by state_capture_filter_gamma::apply_lut(), compress_frame_tiles(), decompress_thread(), task_run_parallel(), and video_reading_thread().

Here is the call graph for this function:

Variable Documentation

instance

class worker_pool instance

static

Definition at line 250 of file worker.cpp.

Referenced by Openxr_session::Openxr_session(), create_filter(), display_xrgl_probe(), wp_worker::enter_loop(), vulkan_display::VulkanDisplay::init(), BasicRTSPOnlyServer::start_server(), task_run_async(), task_run_async_detached(), and wait_task().