task.cc File Reference

Rudimentary, non-preemptive task system in portable C, based on Tom Duff's switch-based coroutine trick and a stack of environment structs. More...

#include "xcom/x_platform.h"
#include "xcom/xcom_memory.h"
#include "xcom/xcom_profile.h"
#include <openssl/err.h>
#include <openssl/ssl.h>
#include <limits.h>
#include <arpa/inet.h>
#include <netdb.h>
#include <assert.h>
#include <errno.h>
#include <fcntl.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <sys/types.h>
#include <time.h>
#include <memory>
#include "xcom/node_connection.h"
#include "xdr_gen/xcom_vp.h"
#include "xcom/simset.h"
#include "xcom/site_def.h"
#include "xcom/task.h"
#include "xcom/task_debug.h"
#include "xcom/task_net.h"
#include "xcom/task_os.h"
#include "xcom/xcom_base.h"
#include "xcom/xcom_cfg.h"
#include "xcom/xcom_transport.h"
#include <poll.h>
#include "xcom/retry.h"
#include "xcom/xdr_utils.h"

Classes
struct	pollfd_array
struct	task_env_p_array
struct	xcom_clock

Macros
#define	xcom_buf   void
#define	crash(x)   g_critical("%s\n", x)
#define	FIX_POS(i)   q->x[i]->heap_pos = (i)
#define	TASK_SWAP(i, j)
#define	TASK_MOVE(i, j)
#define	STAT_INTERVAL   1.0

Typedefs
typedef struct iotasks	iotasks
typedef task_env *	task_env_p
typedef struct xcom_clock	xcom_clock

Functions
const char *	pax_op_to_str (int x)
uint32_t	get_my_xcom_id ()
	init_xdr_array (pollfd) free_xdr_array(pollfd) set_xdr_array(pollfd) get_xdr_array(pollfd) init_xdr_array(task_env_p) free_xdr_array(task_env_p) set_xdr_array(task_env_p) get_xdr_array(task_env_p) struct iotasks
static task_env *	extract_first_delayed ()
static task_env *	task_ref (task_env *t)
static task_env *	task_unref (task_env *t)
static void	wake_all_io ()
static void	task_sys_deinit ()
static double	ts_to_sec (struct timespec *ts)
static double	get_monotonic_time ()
static double	get_real_time ()
static double	xcom_monotonic_seconds (xcom_clock *clock)
static void	xcom_init_clock (xcom_clock *clock)
double	seconds ()
double	task_now ()
static void	task_queue_siftup (task_queue *q, int n)
static void	task_queue_siftdown (task_queue *q, int l, int n)
static task_env *	task_queue_remove (task_queue *q, int i)
static void	task_queue_insert (task_queue *q, task_env *t)
static int	task_queue_empty (task_queue *q)
static task_env *	task_queue_min (task_queue *q)
static task_env *	task_queue_extractmin (task_queue *q)
static void	task_init (task_env *t)
	Initialize task memory. More...
void *	task_allocate (task_env *p, unsigned int bytes)
	Allocate bytes from pool, initialized to zero. More...
static task_env *	activate (task_env *t)
static task_env *	deactivate (task_env *t)
void	task_delay_until (double time)
void	task_wait (task_env *t, linkage *queue)
void	task_wakeup (linkage *queue)
static void	task_wakeup_first (linkage *queue)
channel *	channel_init (channel *c, unsigned int type)
void	channel_put (channel *c, linkage *data)
void	channel_put_front (channel *c, linkage *data)
task_env *	task_new (task_func func, task_arg arg, const char *name, int debug)
void	reset_state (task_env *p)
void	pushp (task_env *p, void *ptr)
void	popp (task_env *p)
static int	runnable_tasks ()
static int	delayed_tasks ()
static void	task_delete (task_env *t)
task_env *	task_activate (task_env *t)
task_env *	task_deactivate (task_env *t)
task_env *	task_terminate (task_env *t)
void	task_terminate_all ()
static task_env *	first_delayed ()
static void	iotasks_init (iotasks *iot_to_init)
static void	iotasks_deinit (iotasks *iot_to_deinit)
static void	poll_wakeup (u_int i)
static int	poll_wait (int ms)
static void	add_fd (task_env *t, int fd, int op)
static void	unpoll (u_int i)
void	remove_and_wakeup (int fd)
task_env *	wait_io (task_env *t, int fd, int op)
result	con_read (connection_descriptor const *rfd, void *buf, int n)
result	con_pipe_read (connection_descriptor const *rfd, void *buf, int n)
int	task_read (connection_descriptor const *con, void *buf, int n, int64_t *ret, connnection_read_method read_function)
result	con_write (connection_descriptor const *wfd, void *buf, int n)
result	con_pipe_write (connection_descriptor const *wfd, void *buf, int n)
int	task_write (connection_descriptor const *con, void *_buf, uint32_t n, int64_t *ret)
int	unblock_fd (int fd)
int	block_fd (int fd)
int	is_only_task ()
static task_env *	first_runnable ()
static task_env *	next_task (task_env *t)
static int	is_task_head (task_env *t)
static int	msdiff (double time)
void	set_should_exit_getter (should_exit_getter x)
void	task_loop ()
static void	init_task_vars ()
void	task_sys_init ()
void	set_task (task_env **p, task_env *t)
const char *	task_name ()

Variables
task_arg	null_arg = {a_end, {0}}
int	task_errno = 0
static xcom_clock	task_timer
static linkage	ash_nazg_gimbatul
static linkage	tasks = {0, &tasks, &tasks}
static task_queue	task_time_q
static linkage	free_tasks = {0, &free_tasks, &free_tasks}
static int	active_tasks = 0
static iotasks	iot
task_env *	stack = nullptr
static uint64_t	send_count
static uint64_t	receive_count
static uint64_t	send_bytes
static uint64_t	receive_bytes
static should_exit_getter	get_should_exit
static double	idle_time = 0.0

Detailed Description

Rudimentary, non-preemptive task system in portable C, based on Tom Duff's switch-based coroutine trick and a stack of environment structs.

(continuations?) Nonblocking IO and event handling need to be rewritten for each new OS. The code is not MT-safe, but could be made safe by moving all global variables into a context struct which could be the first parameter to all the functions.

Macro Definition Documentation

crash

#define crash

(

x

)

g_critical("%s\n", x)

FIX_POS

#define FIX_POS

(

i

)

q->x[i]->heap_pos = (i)

STAT_INTERVAL

#define STAT_INTERVAL   1.0

TASK_MOVE

#define TASK_MOVE	(		i,
			j
	)

Value:

  {                     \
    q->x[i] = q->x[j];  \
    FIX_POS(i);         \
  }

TASK_SWAP

#define TASK_SWAP	(		i,
			j
	)

Value:

  {                          \
    task_env *tmp = q->x[i]; \
    q->x[i] = q->x[j];       \
    q->x[j] = tmp;           \
    FIX_POS(i);              \
    FIX_POS(j);              \
  }

xcom_buf

#define xcom_buf   void

Typedef Documentation

iotasks

typedef struct iotasks iotasks

task_env_p

typedef task_env* task_env_p

xcom_clock

typedef struct xcom_clock xcom_clock

Function Documentation

activate()

static task_env * activate ( task_env *  t )

static

add_fd()

static void add_fd ( task_env *  t, int  fd, int  op  )

static

block_fd()

int block_fd

(

int

fd

)

channel_init()

channel * channel_init	(	channel *	c,
		unsigned int	type
	)

channel_put()

void channel_put	(	channel *	c,
		linkage *	data
	)

channel_put_front()

void channel_put_front	(	channel *	c,
		linkage *	data
	)

con_pipe_read()

result con_pipe_read	(	connection_descriptor const *	rfd,
		void *	buf,
		int	n
	)

con_pipe_write()

result con_pipe_write	(	connection_descriptor const *	wfd,
		void *	buf,
		int	n
	)

con_read()

result con_read	(	connection_descriptor const *	rfd,
		void *	buf,
		int	n
	)

con_write()

result con_write	(	connection_descriptor const *	wfd,
		void *	buf,
		int	n
	)

deactivate()

static task_env * deactivate ( task_env *  t )

static

delayed_tasks()

static int delayed_tasks ( )

static

extract_first_delayed()

static task_env * extract_first_delayed ( )

static

first_delayed()

static task_env * first_delayed ( )

static

first_runnable()

static task_env * first_runnable ( )

static

get_monotonic_time()

static double get_monotonic_time ( )

inlinestatic

get_my_xcom_id()

uint32_t get_my_xcom_id

(

)

get_real_time()

static double get_real_time ( )

inlinestatic

init_task_vars()

static void init_task_vars ( )

static

init_xdr_array()

init_xdr_array

(

pollfd

)

iotasks_deinit()

static void iotasks_deinit ( iotasks *  iot_to_deinit )

static

iotasks_init()

static void iotasks_init ( iotasks *  iot_to_init )

static

is_only_task()

int is_only_task

(

)

is_task_head()

static int is_task_head ( task_env *  t )

static

msdiff()

static int msdiff ( double  time )

static

next_task()

static task_env * next_task ( task_env *  t )

static

pax_op_to_str()

const char * pax_op_to_str

(

int

x

)

poll_wait()

static int poll_wait ( int  ms )

static

poll_wakeup()

static void poll_wakeup ( u_int  i )

static

popp()

void popp

(

task_env *

p

)

pushp()

void pushp	(	task_env *	p,
		void *	ptr
	)

remove_and_wakeup()

void remove_and_wakeup

(

int

fd

)

reset_state()

void reset_state

(

task_env *

p

)

runnable_tasks()

static int runnable_tasks ( )

static

seconds()

double seconds

(

)

set_should_exit_getter()

void set_should_exit_getter

(

should_exit_getter

x

)

set_task()

void set_task	(	task_env **	p,
		task_env *	t
	)

task_activate()

task_env * task_activate

(

task_env *

t

)

task_allocate()

void * task_allocate	(	task_env *	p,
		unsigned int	bytes
	)

Allocate bytes from pool, initialized to zero.

task_deactivate()

task_env * task_deactivate

(

task_env *

t

)

task_delay_until()

void task_delay_until

(

double

time

)

task_delete()

static void task_delete ( task_env *  t )

static

task_init()

static void task_init ( task_env *  p )

static

Initialize task memory.

task_loop()

void task_loop

(

)

task_name()

const char * task_name

(

)

task_new()

task_env * task_new	(	task_func	func,
		task_arg	arg,
		const char *	name,
		int	debug
	)

task_now()

double task_now

(

)

task_queue_empty()

static int task_queue_empty ( task_queue *  q )

static

task_queue_extractmin()

static task_env * task_queue_extractmin ( task_queue *  q )

static

task_queue_insert()

static void task_queue_insert ( task_queue *  q, task_env *  t  )

static

task_queue_min()

static task_env * task_queue_min ( task_queue *  q )

static

task_queue_remove()

static task_env * task_queue_remove ( task_queue *  q, int  i  )

static

task_queue_siftdown()

static void task_queue_siftdown ( task_queue *  q, int  l, int  n  )

static

task_queue_siftup()

static void task_queue_siftup ( task_queue *  q, int  n  )

static

task_read()

int task_read	(	connection_descriptor const *	con,
		void *	buf,
		int	n,
		int64_t *	ret,
		connnection_read_method	read_function
	)

task_ref()

static task_env * task_ref ( task_env *  t )

static

task_sys_deinit()

static void task_sys_deinit ( )

static

task_sys_init()

void task_sys_init

(

)

task_terminate()

task_env * task_terminate

(

task_env *

t

)

task_terminate_all()

void task_terminate_all

(

)

task_unref()

static task_env * task_unref ( task_env *  t )

static

task_wait()

void task_wait	(	task_env *	t,
		linkage *	queue
	)

task_wakeup()

void task_wakeup

(

linkage *

queue

)

task_wakeup_first()

static void task_wakeup_first ( linkage *  queue )

static

task_write()

int task_write	(	connection_descriptor const *	con,
		void *	_buf,
		uint32_t	n,
		int64_t *	ret
	)

ts_to_sec()

static double ts_to_sec ( struct timespec *  ts )

inlinestatic

unblock_fd()

int unblock_fd

(

int

fd

)

unpoll()

static void unpoll ( u_int  i )

static

wait_io()

task_env * wait_io	(	task_env *	t,
		int	fd,
		int	op
	)

wake_all_io()

static void wake_all_io ( )

static

xcom_init_clock()

static void xcom_init_clock ( xcom_clock *  clock )

static

xcom_monotonic_seconds()

static double xcom_monotonic_seconds ( xcom_clock *  clock )

static

Variable Documentation

active_tasks

int active_tasks = 0

static

ash_nazg_gimbatul

linkage ash_nazg_gimbatul

static

Initial value:

= {
    0, &ash_nazg_gimbatul, &ash_nazg_gimbatul}

free_tasks

linkage free_tasks = {0, &free_tasks, &free_tasks}

static

get_should_exit

should_exit_getter get_should_exit

static

idle_time

double idle_time = 0.0

static

iot

iotasks iot

static

null_arg

task_arg null_arg = {a_end, {0}}

receive_bytes

uint64_t receive_bytes

static

receive_count

uint64_t receive_count

static

send_bytes

uint64_t send_bytes

static

send_count

uint64_t send_count

static

stack

task_env* stack = nullptr

task_errno

int task_errno = 0

task_time_q

task_queue task_time_q

static

task_timer

xcom_clock task_timer

static

tasks

linkage tasks = {0, &tasks, &tasks}

static