task.h File Reference

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

#include "xcom/xcom_profile.h"
#include <assert.h>
#include <errno.h>
#include "my_compiler.h"
#include "xcom/simset.h"
#include "xcom/task_arg.h"
#include "xcom/x_platform.h"
#include "xcom/xcom_common.h"
#include "xcom/node_connection.h"
#include "xcom/result.h"

Go to the source code of this file.

Classes
struct	task_ptr
struct	task_env
struct	task_queue
struct	channel

Macros
#define	ADD_EVENTS(x)
#define	ADD_DBG(d, x)
#define	ADD_T_EV(when, file, state, what)
#define	ADD_WAIT_EV(when, file, state, what, milli)
#define	TASK_POOL_ELEMS   1000
#define	MAXTASKS   1000
#define	_ep   ((struct env *)(stack->sp->ptr))
#define	TASK_ALLOC(pool, type)   (task_allocate(pool, (unsigned int)sizeof(type)))
#define	TASK_DEBUG(x)
#define	FINALLY    task_cleanup:
#define	ON_STACK_TOP   (stack->sp == stack->stack_top + 1)
#define	TERM_CHECK    if (ON_STACK_TOP && stack->terminate) goto task_cleanup
#define	TERMINATE   goto task_cleanup
#define	TASK_BEGIN
#define	TASK_END
#define	TASK_RETURN(x)
#define	TASK_DUMP_ERR
#define	TASK_FAIL
#define	TASK_YIELD
#define	TASK_DEACTIVATE
#define	TASK_DELAY(t)
#define	TASK_DELAY_UNTIL(t)
#define	TASK_WAIT(queue)
#define	TIMED_TASK_WAIT(queue, t)
#define	CHANNEL_GET(channel, ptr, type)
#define	CHANNEL_PEEK(channel, ptr, type)
#define	CHANNEL_GET_REVERSE(channel, ptr, type)
#define	TASK_CALL(funcall)
#define	DECL_ENV   struct env {
#define	ENV_INIT   void init() {
#define	END_ENV_INIT   }
#define	END_ENV
#define	LOCK_FD(fd, op)
#define	UNLOCK_FD(fd, op)   unlock_fd(fd, stack, op)
#define	xstr(s)   #s

Typedefs
typedef struct task_ptr	TaskAlign
typedef int(*	task_func) (task_arg arg)
typedef enum terminate_enum	terminate_enum
typedef struct task_env	task_env
typedef struct task_queue	task_queue
typedef struct channel	channel
typedef result(*	connnection_read_method) (connection_descriptor const *rfd, void *buf, int n)
typedef result(*	connnection_write_method) (connection_descriptor const *rfd, void *buf, int n)

Enumerations
enum	terminate_enum { RUN = 0 , KILL = 1 , TERMINATED = 2 }

Functions
static void	set_int_arg (task_arg *arg, int value)
static int	get_int_arg (task_arg arg)
static void	set_long_arg (task_arg *arg, long value)
static long	get_long_arg (task_arg arg)
static void	set_uint_arg (task_arg *arg, unsigned int value)
static unsigned int	get_uint_arg (task_arg arg)
static void	set_ulong_arg (task_arg *arg, unsigned long value)
static void	set_ulong_long_arg (task_arg *arg, unsigned long long value)
static unsigned long	get_ulong_arg (task_arg arg)
static unsigned long long	get_ulong_long_arg (task_arg arg)
static void	set_float_arg (task_arg *arg, float value)
static float	get_float_arg (task_arg arg)
static void	set_double_arg (task_arg *arg, double value)
static double	get_double_arg (task_arg arg)
static void	set_string_arg (task_arg *arg, char const *value)
static void	set_void_arg (task_arg *arg, void *value)
static char const *	get_string_arg (task_arg arg)
static void *	get_void_arg (task_arg arg)
static task_arg	int_arg (int i)
static task_arg	uint_arg (unsigned int i)
static task_arg	ulong_arg (unsigned long l)
static task_arg	ulong_long_arg (unsigned long long ll)
static task_arg	double_arg (double i)
static task_arg	string_arg (char const *v)
static task_arg	void_arg (void *v)
static task_arg	end_arg ()
channel *	channel_init (channel *c, unsigned int type)
void	channel_put (channel *c, linkage *data)
void	channel_put_front (channel *c, linkage *data)
void *	task_allocate (task_env *p, unsigned int bytes)
	Allocate bytes from pool, initialized to zero. More...
void	reset_state (task_env *p)
void	pushp (task_env *p, void *ptr)
void	popp (task_env *p)
double	seconds ()
double	task_now ()
void	task_delay_until (double time)
unsigned long long int	get_time_since_the_epoch ()
	Return time in microseconds. More...
int	unblock_fd (int fd)
int	block_fd (int fd)
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=con_read)
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	is_locked (int fd)
int	lock_fd (int fd, task_env *t, int lock)
int	unlock_fd (int fd, task_env *t, int lock)
void	task_sys_init ()
task_env *	task_new (task_func func, task_arg arg, const char *name, int debug)
void	task_loop ()
void	task_wait (task_env *t, linkage *queue)
void	task_wakeup (linkage *queue)
task_env *	task_terminate (task_env *t)
void	set_task (task_env **p, task_env *t)
void	task_terminate_all ()
void	remove_and_wakeup (int i)
int	is_only_task ()
task_env *	task_activate (task_env *t)
task_env *	task_deactivate (task_env *t)
const char *	task_name ()
task_env *	wait_io (task_env *t, int fd, int op)
result	set_nodelay (int fd)
task_env *	timed_wait_io (task_env *t, int fd, int op, double timeout)

Variables
task_env *	stack
int	task_errno
xcom_proto const	my_min_xcom_version
xcom_proto const	my_xcom_version

Detailed Description

Rudimentary 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.

Macro Definition Documentation

_ep

#define _ep   ((struct env *)(stack->sp->ptr))

ADD_DBG

#define ADD_DBG	(		d,
			x
	)

ADD_EVENTS

#define ADD_EVENTS

(

x

)

ADD_T_EV

#define ADD_T_EV	(		when,
			file,
			state,
			what
	)

ADD_WAIT_EV

#define ADD_WAIT_EV	(		when,
			file,
			state,
			what,
			milli
	)

CHANNEL_GET

#define CHANNEL_GET	(		channel,
			ptr,
			type
	)

Value:

  {                                                                \
    while (link_empty(&(channel)->data)) {                         \
      TASK_WAIT(&(channel)->queue);                                \
    }                                                              \
    *(ptr) = (type *)link_extract_first(&(channel)->data);         \
    IFDBG(D_TRANSPORT, FN; STRLIT("CHANNEL_GET "); PTREXP(*(ptr)); \
          PTREXP(&((channel)->data)));                             \
  }

CHANNEL_GET_REVERSE

#define CHANNEL_GET_REVERSE	(		channel,
			ptr,
			type
	)

Value:

  {                                                       \
    while (link_empty(&(channel)->data)) {                \
      TASK_WAIT(&(channel)->queue);                       \
    }                                                     \
    *(ptr) = (type *)link_extract_last(&(channel)->data); \
  }

CHANNEL_PEEK

#define CHANNEL_PEEK	(		channel,
			ptr,
			type
	)

Value:

  {                                                \
    while (link_empty(&(channel)->data)) {         \
      TASK_WAIT(&(channel)->queue);                \
    }                                              \
    *(ptr) = (type *)link_first(&(channel)->data); \
  }

DECL_ENV

#define DECL_ENV   struct env {

END_ENV

#define END_ENV

Value:

  }             \
  ;             \
  [[maybe_unused]] struct env *ep

END_ENV_INIT

#define END_ENV_INIT   }

ENV_INIT

#define ENV_INIT   void init() {

FINALLY

#define FINALLY    task_cleanup:

LOCK_FD

#define LOCK_FD	(		fd,
			op
	)

Value:

  {                                                                            \
    while (!lock_fd(                                                           \
        fd, stack,                                                             \
        op)) { /* Effectively a spin lock, but should not happen very often */ \
      wait_io(stack, fd, op);                                                  \
      TASK_YIELD;                                                              \
      /* TASK_DELAY(1.0);     */                                               \
    }                                                                          \
  }

MAXTASKS

#define MAXTASKS   1000

ON_STACK_TOP

#define ON_STACK_TOP   (stack->sp == stack->stack_top + 1)

TASK_ALLOC

#define TASK_ALLOC	(		pool,
			type
	)		(task_allocate(pool, (unsigned int)sizeof(type)))

TASK_BEGIN

#define TASK_BEGIN

Value:

  /* assert(ep);   */                                         \
  ADD_DBG(                                                    \
      D_TASK,                                                 \
      if (stack->sp->state) {                                 \
        add_event(EVENT_DUMP_PAD, string_arg("state"));       \
        add_event(EVENT_DUMP_PAD, int_arg(stack->sp->state)); \
      } add_event(EVENT_DUMP_PAD, string_arg("TASK_BEGIN"));  \
      add_event(EVENT_DUMP_PAD, void_arg(stack)););           \
  TASK_DEBUG("TASK_BEGIN");                                   \
  switch (stack->sp->state) {                                 \
    case 0:                                                   \
      pushp(stack, TASK_ALLOC(stack, struct env));            \
      ep = _ep;                                               \
      assert(ep);                                             \
      ep->init();                                             \
      TERM_CHECK;

TASK_CALL

#define TASK_CALL

(

funcall

)

Value:

  {                                   \
    reset_state(stack);               \
    TASK_DEBUG("BEFORE CALL");        \
    do {                              \
      stack->sp--;                    \
      stack->taskret = funcall;       \
      stack->sp++;                    \
      TERM_CHECK;                     \
      if (stack->taskret) TASK_YIELD; \
    } while (stack->taskret);         \
    TASK_DEBUG("AFTER CALL");         \
  }

TASK_DEACTIVATE

#define TASK_DEACTIVATE

Value:

  {                                \
    TASK_DEBUG("TASK_DEACTIVATE"); \
    task_deactivate(stack);        \
    TASK_YIELD;                    \
  }

TASK_DEBUG

#define TASK_DEBUG

(

x

)

TASK_DELAY

#define TASK_DELAY

(

t

)

Value:

  {                                  \
    TASK_DEBUG("TASK_DELAY");        \
    task_delay_until(seconds() + t); \
    TASK_YIELD;                      \
  }

TASK_DELAY_UNTIL

#define TASK_DELAY_UNTIL

(

t

)

Value:

  {                                 \
    TASK_DEBUG("TASK_DELAY_UNTIL"); \
    task_delay_until(t);            \
    TASK_YIELD;                     \
  }

TASK_DUMP_ERR

#define TASK_DUMP_ERR

Value:

  if (errno || SOCK_ERRNO || task_errno) {                     \
    IFDBG(D_NONE, FN; NDBG(errno, d); STREXP(strerror(errno)); \
          NDBG(SOCK_ERRNO, d); STREXP(strerror(SOCK_ERRNO));   \
          NDBG(task_errno, d); STREXP(strerror(task_errno)));  \
  }

TASK_END

#define TASK_END

Value:

  ADD_DBG(                                                    \
      D_TASK,                                                 \
      if (stack->sp->state) {                                 \
        add_event(EVENT_DUMP_PAD, string_arg("state"));       \
        add_event(EVENT_DUMP_PAD, int_arg(stack->sp->state)); \
      } add_event(EVENT_DUMP_PAD, string_arg("TASK_END"));    \
      add_event(EVENT_DUMP_PAD, void_arg(stack)););           \
  TASK_DEBUG("TASK_END");                                     \
  stack->sp->state = 0;                                       \
  stack->where = (TaskAlign *)stack->sp->ptr;                 \
  assert(stack->where);                                       \
  popp(stack);                                                \
  return 0;                                                   \
  }                                                           \
  return 0

TASK_FAIL

#define TASK_FAIL

Value:

  {                                                                        \
    *ret = (-1);                                                           \
    TASK_DUMP_ERR;                                                         \
    IFDBG(D_NONE, FN; STRLIT("TASK_FAIL"));                                \
    ADD_DBG(D_TASK, add_event(EVENT_DUMP_PAD, string_arg("task failed"))); \
    goto task_cleanup;                                                     \
  }

TASK_POOL_ELEMS

#define TASK_POOL_ELEMS   1000

TASK_RETURN

#define TASK_RETURN

(

x

)

Value:

  {                    \
    *ret = (x);        \
    goto task_cleanup; \
  }

TASK_WAIT

#define TASK_WAIT

(

queue

)

Value:

  {                          \
    TASK_DEBUG("TASK_WAIT"); \
    task_wait(stack, queue); \
    TASK_YIELD;              \
  }

TASK_YIELD

#define TASK_YIELD

Value:

  {                                    \
    TASK_DEBUG("TASK_YIELD");          \
    stack->sp->state = __LINE__;       \
    return 1;                          \
    case __LINE__:                     \
      TASK_DEBUG("RETURN FROM YIELD"); \
      ep = _ep;                        \
      assert(ep);                      \
      TERM_CHECK;                      \
  }

TERM_CHECK

#define TERM_CHECK    if (ON_STACK_TOP && stack->terminate) goto task_cleanup

TERMINATE

#define TERMINATE   goto task_cleanup

TIMED_TASK_WAIT

#define TIMED_TASK_WAIT	(		queue,
			t
	)

Value:

  {                                    \
    TASK_DEBUG("TIMED_TASK_WAIT");     \
    task_delay_until(seconds() + (t)); \
    task_wait(stack, queue);           \
    TASK_YIELD;                        \
  }

UNLOCK_FD

#define UNLOCK_FD	(		fd,
			op
	)		unlock_fd(fd, stack, op)

xstr

#define xstr

(

s

)

#s

Typedef Documentation

channel

typedef struct channel channel

connnection_read_method

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

connnection_write_method

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

task_env

typedef struct task_env task_env

task_func

typedef int(* task_func) (task_arg arg)

task_queue

typedef struct task_queue task_queue

TaskAlign

typedef struct task_ptr TaskAlign

terminate_enum

typedef enum terminate_enum terminate_enum

Enumeration Type Documentation

terminate_enum

enum terminate_enum

Enumerator
RUN
KILL
TERMINATED

Function Documentation

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
	)

double_arg()

static task_arg double_arg ( double  i )

inlinestatic

end_arg()

static task_arg end_arg ( )

inlinestatic

get_double_arg()

static double get_double_arg ( task_arg  arg )

inlinestatic

get_float_arg()

static float get_float_arg ( task_arg  arg )

inlinestatic

get_int_arg()

static int get_int_arg ( task_arg  arg )

inlinestatic

get_long_arg()

static long get_long_arg ( task_arg  arg )

inlinestatic

get_string_arg()

static char const * get_string_arg ( task_arg  arg )

inlinestatic

get_time_since_the_epoch()

unsigned long long int get_time_since_the_epoch

(

)

Return time in microseconds.

Uses std::chrono::high_resolution_clock

Return values

Number

of microseconds since the Epoch, 1970-01-01 00:00:00 +0000 (UTC)

get_uint_arg()

static unsigned int get_uint_arg ( task_arg  arg )

inlinestatic

get_ulong_arg()

static unsigned long get_ulong_arg ( task_arg  arg )

inlinestatic

get_ulong_long_arg()

static unsigned long long get_ulong_long_arg ( task_arg  arg )

inlinestatic

get_void_arg()

static void * get_void_arg ( task_arg  arg )

inlinestatic

int_arg()

static task_arg int_arg ( int  i )

inlinestatic

is_locked()

int is_locked

(

int

fd

)

is_only_task()

int is_only_task

(

)

lock_fd()

int lock_fd	(	int	fd,
		task_env *	t,
		int	lock
	)

popp()

void popp

(

task_env *

p

)

pushp()

void pushp	(	task_env *	p,
		void *	ptr
	)

remove_and_wakeup()

void remove_and_wakeup

(

int

i

)

reset_state()

void reset_state

(

task_env *

p

)

seconds()

double seconds

(

)

set_double_arg()

static void set_double_arg ( task_arg *  arg, double  value  )

inlinestatic

set_float_arg()

static void set_float_arg ( task_arg *  arg, float  value  )

inlinestatic

set_int_arg()

static void set_int_arg ( task_arg *  arg, int  value  )

inlinestatic

set_long_arg()

static void set_long_arg ( task_arg *  arg, long  value  )

inlinestatic

set_nodelay()

result set_nodelay

(

int

fd

)

set_string_arg()

static void set_string_arg ( task_arg *  arg, char const *  value  )

inlinestatic

set_task()

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

set_uint_arg()

static void set_uint_arg ( task_arg *  arg, unsigned int  value  )

inlinestatic

set_ulong_arg()

static void set_ulong_arg ( task_arg *  arg, unsigned long  value  )

inlinestatic

set_ulong_long_arg()

static void set_ulong_long_arg ( task_arg *  arg, unsigned long long  value  )

inlinestatic

set_void_arg()

static void set_void_arg ( task_arg *  arg, void *  value  )

inlinestatic

string_arg()

static task_arg string_arg ( char const *  v )

inlinestatic

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_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_read()

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

task_sys_init()

void task_sys_init

(

)

task_terminate()

task_env * task_terminate

(

task_env *

t

)

task_terminate_all()

void task_terminate_all

(

)

task_wait()

void task_wait	(	task_env *	t,
		linkage *	queue
	)

task_wakeup()

void task_wakeup

(

linkage *

queue

)

task_write()

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

timed_wait_io()

task_env * timed_wait_io	(	task_env *	t,
		int	fd,
		int	op,
		double	timeout
	)

uint_arg()

static task_arg uint_arg ( unsigned int  i )

inlinestatic

ulong_arg()

static task_arg ulong_arg ( unsigned long  l )

inlinestatic

ulong_long_arg()

static task_arg ulong_long_arg ( unsigned long long  ll )

inlinestatic

unblock_fd()

int unblock_fd

(

int

fd

)

unlock_fd()

int unlock_fd	(	int	fd,
		task_env *	t,
		int	lock
	)

void_arg()

static task_arg void_arg ( void *  v )

inlinestatic

wait_io()

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

Variable Documentation

my_min_xcom_version

xcom_proto const my_min_xcom_version

extern

my_xcom_version

xcom_proto const my_xcom_version

extern

stack

task_env* stack

extern

task_errno

int task_errno

extern