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"

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

Classes

Macros

Typedefs

Enumerations

Functions

Variables

Detailed Description

Macro Definition Documentation

◆ _ep

◆ ADD_DBG

◆ ADD_EVENTS

◆ ADD_T_EV

◆ ADD_WAIT_EV

◆ CHANNEL_GET

◆ CHANNEL_GET_REVERSE

◆ CHANNEL_PEEK

◆ DECL_ENV

◆ END_ENV

◆ END_ENV_INIT

◆ ENV_INIT

◆ FINALLY

◆ LOCK_FD

◆ MAXTASKS

◆ ON_STACK_TOP

◆ TASK_ALLOC

◆ TASK_BEGIN

◆ TASK_CALL

◆ TASK_DEACTIVATE

◆ TASK_DEBUG

◆ TASK_DELAY

◆ TASK_DELAY_UNTIL

◆ TASK_DUMP_ERR

◆ TASK_END

◆ TASK_FAIL

◆ TASK_POOL_ELEMS

◆ TASK_RETURN

◆ TASK_WAIT

◆ TASK_YIELD

◆ TERM_CHECK

◆ TERMINATE

◆ TIMED_TASK_WAIT

◆ UNLOCK_FD

◆ xstr

Typedef Documentation

◆ channel

◆ connnection_read_method

◆ connnection_write_method

◆ task_env

◆ task_func

◆ task_queue

◆ TaskAlign

◆ terminate_enum

Enumeration Type Documentation

◆ terminate_enum

Function Documentation

◆ block_fd()

◆ channel_init()

◆ channel_put()

◆ channel_put_front()

◆ con_pipe_read()

◆ con_pipe_write()

◆ con_read()

◆ con_write()

◆ double_arg()

◆ end_arg()

◆ get_double_arg()

◆ get_float_arg()

◆ get_int_arg()

◆ get_long_arg()

◆ get_string_arg()

◆ get_time_since_the_epoch()

◆ get_uint_arg()

◆ get_ulong_arg()

◆ get_ulong_long_arg()

◆ get_void_arg()

◆ int_arg()

◆ is_locked()

◆ is_only_task()

◆ lock_fd()

◆ popp()