Mirrors/libwebsockets
1
0
Fork 0
mirror of https://github.com/warmcat/libwebsockets.git synced 2025-03-09 00:00:04 +01:00
Code Issues Releases Wiki Activity

sequencer: upgrade timeout to use us

Browse source 
Adapt service loops and event libs to use microsecond waits
internally, for hrtimer and sequencer.  Reduce granularity
according to platform / event lib wait.

Add a helper so there's a single place to extend it.
This commit is contained in:
Andy Green 2019-08-08 06:30:14 +01:00
parent c27c38ffbc
commit fed78bef42
24 changed files with 286 additions and 189 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

5
CMakeLists.txt
View file

@ -1305,6 +1305,11 @@ if (LWS_WITH_HTTP_PROXY)
lib/roles/http/server/rewrite.c)
endif()
if (LWS_WITH_NETWORK)
list(APPEND SOURCES
lib/event-libs/common.c)
endif()
if (LWS_WITH_POLL AND LWS_WITH_NETWORK)
list(APPEND SOURCES
lib/event-libs/poll/poll.c)

6
include/libwebsockets.h
View file

@ -42,6 +42,12 @@ extern "C" {
* CARE: everything using cmake defines needs to be below here
*/
#define LWS_US_PER_SEC 1000000
#define LWS_MS_PER_SEC 1000
#define LWS_US_PER_MS 1000
#define LWS_US_TO_MS(x) ((x + (LWS_US_PER_MS / 2)) / LWS_US_PER_MS)
#if defined(LWS_HAS_INTPTR_T)
#include <stdint.h>
#define lws_intptr_t intptr_t

2
include/libwebsockets/abstract/abstract.h
View file

@ -73,7 +73,7 @@ typedef struct lws_abs {
const struct lws_abs_transport *at;
const lws_token_map_t *at_tokens;
lws_sequencer_t *seq;
lws_seq_t *seq;
void *opaque_user_data;
/*

8
include/libwebsockets/lws-client.h
View file

@ -49,7 +49,7 @@ enum lws_client_connect_ssl_connection_flags {
* */
};
typedef struct lws_sequencer lws_sequencer_t;
typedef struct lws_sequencer lws_seq_t;
/** struct lws_client_connect_info - parameters to connect with when using
* lws_client_connect_via_info() */
@ -121,8 +121,8 @@ struct lws_client_connect_info {
* tokens
*/
lws_sequencer_t *seq;
/**< NULL, or an lws_sequencer_t that wants to be given messages about
lws_seq_t *seq;
/**< NULL, or an lws_seq_t that wants to be given messages about
* this wsi's lifecycle as it connects, errors or closes.
*/
@ -130,7 +130,7 @@ struct lws_client_connect_info {
/**< This data has no meaning to lws but is applied to the client wsi
* and can be retrieved by user code with lws_get_opaque_user_data().
* It's also provided with sequencer messages if the wsi is bound to
* an lws_sequencer_t.
* an lws_seq_t.
*/
/* Add new things just above here ---^

72
include/libwebsockets/lws-sequencer.h
View file

@ -56,7 +56,7 @@ typedef enum lws_seq_cb_return {
LWSSEQ_RET_DESTROY
} lws_seq_cb_return_t;
typedef struct lws_sequencer lws_sequencer_t; /* opaque */
typedef struct lws_sequencer lws_seq_t; /* opaque */
/*
* handler for this sequencer. Return 0 if OK else nonzero to destroy the
@ -83,7 +83,7 @@ typedef struct lws_seq_info {
} lws_seq_info_t;
/**
* lws_sequencer_create() - create and bind sequencer to a pt
* lws_seq_create() - create and bind sequencer to a pt
*
* \param info: information about sequencer to create
*
@ -97,26 +97,26 @@ typedef struct lws_seq_info {
*
* pt locking is used to protect the related data structures.
*/
LWS_VISIBLE LWS_EXTERN lws_sequencer_t *
lws_sequencer_create(lws_seq_info_t *info);
LWS_VISIBLE LWS_EXTERN lws_seq_t *
lws_seq_create(lws_seq_info_t *info);
/**
* lws_sequencer_destroy() - destroy the sequencer
* lws_seq_destroy() - destroy the sequencer
*
* \param seq: pointer to the the opaque sequencer pointer returned by
* lws_sequencer_create()
* lws_seq_create()
*
* This proceeds to destroy the sequencer, calling LWSSEQ_DESTROYED and then
* freeing the sequencer object itself. The pointed-to seq pointer will be
* set to NULL.
*/
LWS_VISIBLE LWS_EXTERN void
lws_sequencer_destroy(lws_sequencer_t **seq);
lws_seq_destroy(lws_seq_t **seq);
/**
* lws_sequencer_queue_event() - queue an event on the given sequencer
* lws_seq_queue_event() - queue an event on the given sequencer
*
* \param seq: the opaque sequencer pointer returned by lws_sequencer_create()
* \param seq: the opaque sequencer pointer returned by lws_seq_create()
* \param e: the event index to queue
* \param data: associated opaque (to lws) data to provide the callback
* \param aux: second opaque data to provide the callback
@ -131,11 +131,11 @@ lws_sequencer_destroy(lws_sequencer_t **seq);
* values here.
*/
LWS_VISIBLE LWS_EXTERN int
lws_sequencer_queue_event(lws_sequencer_t *seq, lws_seq_events_t e, void *data,
lws_seq_queue_event(lws_seq_t *seq, lws_seq_events_t e, void *data,
void *aux);
/**
* lws_sequencer_check_wsi() - check if wsi still extant
* lws_seq_check_wsi() - check if wsi still extant
*
* \param seq: the sequencer interested in the wsi
* \param wsi: the wsi we want to confirm hasn't closed yet
@ -151,15 +151,17 @@ lws_sequencer_queue_event(lws_sequencer_t *seq, lws_seq_events_t e, void *data,
* close message yet.
*/
LWS_VISIBLE LWS_EXTERN int
lws_sequencer_check_wsi(lws_sequencer_t *seq, struct lws *wsi);
lws_seq_check_wsi(lws_seq_t *seq, struct lws *wsi);
#define LWSSEQTO_NONE 0
/**
* lws_sequencer_timeout() - set a timeout by which the sequence must have
* completed by a different event or inform the
* sequencer
* lws_seq_timeout_us() - set a timeout by which the sequence must have
* completed by a different event or inform the
* sequencer
*
* \param seq: The sequencer to set the timeout on
* \param secs: How many seconds in the future to fire the timeout (0 = disable)
* \param us: How many us in the future to fire the timeout (0 = disable)
*
* This api allows the sequencer to ask to be informed if it has not completed
* or disabled its timeout after secs seconds. Lws will send a LWSSEQ_TIMED_OUT
@ -178,54 +180,54 @@ lws_sequencer_check_wsi(lws_sequencer_t *seq, struct lws *wsi);
* react appropriately.
*/
LWS_VISIBLE LWS_EXTERN int
lws_sequencer_timeout(lws_sequencer_t *seq, int secs);
lws_seq_timeout_us(lws_seq_t *seq, lws_usec_t us);
/**
* lws_sequencer_from_user(): get the lws_sequencer_t pointer from the user ptr
* lws_seq_from_user(): get the lws_seq_t pointer from the user ptr
*
* \param u: the sequencer user allocation returned by lws_sequencer_create() or
* \param u: the sequencer user allocation returned by lws_seq_create() or
* provided in the sequencer callback
*
* This gets the lws_sequencer_t * from the sequencer user allocation pointer.
* This gets the lws_seq_t * from the sequencer user allocation pointer.
* Actually these are allocated at the same time in one step, with the user
* allocation immediately after the lws_sequencer_t, so lws can compute where
* the lws_sequencer_t is from having the user allocation pointer. Since the
* size of the lws_sequencer_t is unknown to user code, this helper does it for
* allocation immediately after the lws_seq_t, so lws can compute where
* the lws_seq_t is from having the user allocation pointer. Since the
* size of the lws_seq_t is unknown to user code, this helper does it for
* you.
*/
LWS_VISIBLE LWS_EXTERN lws_sequencer_t *
lws_sequencer_from_user(void *u);
LWS_VISIBLE LWS_EXTERN lws_seq_t *
lws_seq_from_user(void *u);
/**
* lws_sequencer_secs_since_creation(): elapsed seconds since sequencer created
* lws_seq_secs_since_creation(): elapsed seconds since sequencer created
*
* \param seq: pointer to the lws_sequencer_t
* \param seq: pointer to the lws_seq_t
*
* Returns the number of seconds elapsed since the lws_sequencer_t was
* Returns the number of seconds elapsed since the lws_seq_t was
* created. This is useful to calculate sequencer timeouts for the current
* step considering a global sequencer lifetime limit.
*/
LWS_VISIBLE LWS_EXTERN int
lws_sequencer_secs_since_creation(lws_sequencer_t *seq);
lws_seq_secs_since_creation(lws_seq_t *seq);
/**
* lws_sequencer_name(): get the name of this sequencer
* lws_seq_name(): get the name of this sequencer
*
* \param seq: pointer to the lws_sequencer_t
* \param seq: pointer to the lws_seq_t
*
* Returns the name given when the sequencer was created. This is useful to
* annotate logging when then are multiple sequencers in play.
*/
LWS_VISIBLE LWS_EXTERN const char *
lws_sequencer_name(lws_sequencer_t *seq);
lws_seq_name(lws_seq_t *seq);
/**
* lws_sequencer_get_context(): get the lws_context sequencer was created on
* lws_seq_get_context(): get the lws_context sequencer was created on
*
* \param seq: pointer to the lws_sequencer_t
* \param seq: pointer to the lws_seq_t
*
* Returns the lws_context. Saves you having to store it if you have a seq
* pointer handy.
*/
LWS_VISIBLE LWS_EXTERN struct lws_context *
lws_sequencer_get_context(lws_sequencer_t *seq);
lws_seq_get_context(lws_seq_t *seq);

14
lib/abstract/test-sequencer.c
View file

@ -21,7 +21,7 @@
*
* A helper for running multiple unit tests against abstract protocols.
*
* An lws_sequencer_t is used to base its actions in the event loop and manage
* An lws_seq_t is used to base its actions in the event loop and manage
* the sequencing of multiple tests. A new abstract connection is instantiated
* for each test using te
*/
@ -35,7 +35,7 @@ struct lws_seq_test_sequencer {
struct lws_context *context;
struct lws_vhost *vhost;
lws_sequencer_t *unit_test_seq;
lws_seq_t *unit_test_seq;
/* holds the per-test token for the unit-test transport to consume */
lws_token_map_t uttt[4];
@ -88,7 +88,7 @@ unit_test_result_cb(const void *cb_user, int disposition)
return -1;
}
lws_sequencer_queue_event(s->unit_test_seq, r, NULL, NULL);
lws_seq_queue_event(s->unit_test_seq, r, NULL, NULL);
((struct lws_seq_test_sequencer *)s)->instance = NULL;
@ -115,7 +115,7 @@ test_sequencer_cb(struct lws_sequencer *seq, void *user, int event, void *data,
switch ((int)event) {
case LWSSEQ_CREATED: /* our sequencer just got started */
lwsl_notice("%s: %s: created\n", __func__,
lws_sequencer_name(seq));
lws_seq_name(seq));
s->state = 0; /* first thing we'll do is the first url */
goto step;
@ -178,7 +178,7 @@ pass:
s->args.results[s->state] = LPE_SUCCEEDED;
done:
lws_sequencer_timeout(lws_sequencer_from_user(s), 0);
lws_seq_timeout_us(lws_seq_from_user(s), LWSSEQTO_NONE);
s->state++;
step:
if (!s->args.tests[s->state].name) {
@ -233,7 +233,7 @@ int
lws_abs_unit_test_sequencer(const lws_test_sequencer_args_t *args)
{
struct lws_seq_test_sequencer *s;
lws_sequencer_t *seq;
lws_seq_t *seq;
lws_seq_info_t i;
memset(&i, 0, sizeof(i));
@ -247,7 +247,7 @@ lws_abs_unit_test_sequencer(const lws_test_sequencer_args_t *args)
* Create a sequencer in the event loop to manage the tests
*/
seq = lws_sequencer_create(&i);
seq = lws_seq_create(&i);
if (!seq) {
lwsl_err("%s: unable to create sequencer\n", __func__);
return 1;

6
lib/abstract/transports/raw-skt.c
View file

@ -97,7 +97,7 @@ callback_abs_client_raw_skt(struct lws *wsi, enum lws_callback_reasons reason,
* our lifecycle events
*/
lws_sequencer_queue_event(wsi->seq, LWSSEQ_WSI_CONNECTED,
lws_seq_queue_event(wsi->seq, LWSSEQ_WSI_CONNECTED,
wsi, NULL);
break;
@ -112,7 +112,7 @@ callback_abs_client_raw_skt(struct lws *wsi, enum lws_callback_reasons reason,
* our lifecycle events
*/
lws_sequencer_queue_event(wsi->seq, LWSSEQ_WSI_CONN_FAIL,
lws_seq_queue_event(wsi->seq, LWSSEQ_WSI_CONN_FAIL,
wsi, NULL);
goto close_path;
@ -128,7 +128,7 @@ callback_abs_client_raw_skt(struct lws *wsi, enum lws_callback_reasons reason,
* our lifecycle events
*/
lws_sequencer_queue_event(wsi->seq, LWSSEQ_WSI_CONN_CLOSE,
lws_seq_queue_event(wsi->seq, LWSSEQ_WSI_CONN_CLOSE,
wsi, NULL);
close_path:

26
lib/abstract/transports/unit-test.c
View file

@ -33,7 +33,7 @@ typedef struct lws_abstxp_unit_test_priv {
char note[128];
struct lws_abs *abs;
lws_sequencer_t *seq;
lws_seq_t *seq;
lws_unit_test_t *current_test;
lws_unit_test_packet_t *expect;
lws_unit_test_packet_test_cb result_cb;
@ -78,7 +78,7 @@ lws_unit_test_packet_dispose(abs_unit_test_priv_t *priv,
priv->disposition = disp;
lws_sequencer_queue_event(priv->seq, UTSEQ_MSG_DISPOSITION_KNOWN,
lws_seq_queue_event(priv->seq, UTSEQ_MSG_DISPOSITION_KNOWN,
NULL, NULL);
return disp;
@ -135,10 +135,10 @@ unit_test_sequencer_cb(struct lws_sequencer *seq, void *user, int event,
switch ((int)event) {
case LWSSEQ_CREATED: /* our sequencer just got started */
lwsl_notice("%s: %s: created\n", __func__,
lws_sequencer_name(seq));
lws_seq_name(seq));
if (s->ai->at->client_conn(s->ai)) {
lwsl_notice("%s: %s: abstract client conn failed\n",
__func__, lws_sequencer_name(seq));
__func__, lws_seq_name(seq));
return LWSSEQ_RET_DESTROY;
}
@ -257,7 +257,8 @@ ph:
goto done;
done:
lws_sequencer_timeout(lws_sequencer_from_user(s), 0);
lws_seq_timeout_us(lws_seq_from_user(s),
LWSSEQTO_NONE);
priv->expect++;
if (!priv->expect->buffer) {
/* the sequence has completed */
@ -281,7 +282,7 @@ lws_atcut_close(lws_abs_transport_inst_t *ati)
lwsl_notice("%s\n", __func__);
lws_sequencer_queue_event(priv->seq, UTSEQ_MSG_CLOSING, NULL, NULL);
lws_seq_queue_event(priv->seq, UTSEQ_MSG_CLOSING, NULL, NULL);
return 0;
}
@ -333,7 +334,7 @@ lws_atcut_tx(lws_abs_transport_inst_t *ati, uint8_t *buf, size_t len)
priv->expect++;
lws_sequencer_queue_event(priv->seq, UTSEQ_MSG_POST_TX_KICK, NULL, NULL);
lws_seq_queue_event(priv->seq, UTSEQ_MSG_POST_TX_KICK, NULL, NULL);
return 0;
}
@ -378,12 +379,13 @@ lws_atcut_client_conn(const lws_abs_t *abs)
priv->disposition = LPE_CONTINUE;
priv->note[0] = '\0';
lws_sequencer_timeout(priv->seq, priv->current_test->max_secs);
lws_seq_timeout_us(priv->seq, priv->current_test->max_secs *
LWS_US_PER_SEC);
lwsl_notice("%s: %s: test '%s': start\n", __func__, abs->ap->name,
priv->current_test->name);
lws_sequencer_queue_event(priv->seq, UTSEQ_MSG_CONNECTING, NULL, NULL);
lws_seq_queue_event(priv->seq, UTSEQ_MSG_CONNECTING, NULL, NULL);
return 0;
}
@ -402,7 +404,7 @@ lws_atcut_ask_for_writeable(lws_abs_transport_inst_t *ati)
* until we have returned to the event loop, just like a real
* callback_on_writable()
*/
lws_sequencer_queue_event(priv->seq, UTSEQ_MSG_WRITEABLE, NULL, NULL);
lws_seq_queue_event(priv->seq, UTSEQ_MSG_WRITEABLE, NULL, NULL);
return 0;
}
@ -415,7 +417,7 @@ static int
lws_atcut_create(lws_abs_t *ai)
{
abs_unit_test_priv_t *priv;
lws_sequencer_t *seq;
lws_seq_t *seq;
lws_seq_info_t i;
seq_priv_t *s;
@ -430,7 +432,7 @@ lws_atcut_create(lws_abs_t *ai)
* Create the sequencer for the steps in a single unit test
*/
seq = lws_sequencer_create(&i);
seq = lws_seq_create(&i);
if (!seq) {
lwsl_err("%s: unable to create sequencer\n", __func__);

12
lib/core-net/private.h
View file

@ -288,6 +288,8 @@ struct lws_context_per_thread {
struct lws_pt_tls tls;
#endif
lws_usec_t last_heartbeat;
struct lws_pollfd *fds;
volatile struct lws_foreign_thread_pollfd * volatile foreign_pfd_list;
#ifdef _WIN32
@ -512,7 +514,7 @@ struct lws {
const struct lws_protocols *protocol;
struct lws_dll same_vh_protocol;
lws_sequencer_t *seq; /* associated sequencer if any */
lws_seq_t *seq; /* associated sequencer if any */
struct lws_dll dll_timeout;
struct lws_dll dll_hrtimer;
@ -921,8 +923,8 @@ lws_issue_raw(struct lws *wsi, unsigned char *buf, size_t len);
LWS_EXTERN void
lws_remove_from_timeout_list(struct lws *wsi);
LWS_EXTERN int
lws_sequencer_timeout_check(struct lws_context_per_thread *pt, time_t now);
LWS_EXTERN lws_usec_t
__lws_seq_timeout_check(struct lws_context_per_thread *pt, lws_usec_t usnow);
LWS_EXTERN struct lws * LWS_WARN_UNUSED_RESULT
lws_client_connect_2(struct lws *wsi);
@ -1040,7 +1042,7 @@ int
lws_pt_do_pending_sequencer_events(struct lws_context_per_thread *pt);
void
lws_sequencer_destroy_all_on_pt(struct lws_context_per_thread *pt);
lws_seq_destroy_all_on_pt(struct lws_context_per_thread *pt);
LWS_EXTERN int
lws_broadcast(struct lws_context *context, int reason, void *in, size_t len);
@ -1094,7 +1096,7 @@ void
__lws_remove_from_timeout_list(struct lws *wsi);
lws_usec_t
__lws_hrtimer_service(struct lws_context_per_thread *pt);
__lws_hrtimer_service(struct lws_context_per_thread *pt, lws_usec_t t);
int

100
lib/core-net/sequencer.c
View file

@ -46,19 +46,19 @@ typedef struct lws_sequencer {
const char *name;
const lws_retry_bo_t *retry;
time_t time_created;
time_t timeout; /* 0 or time we timeout */
lws_usec_t time_created;
lws_usec_t timeout; /* 0 or time we timeout */
char going_down;
} lws_sequencer_t;
} lws_seq_t;
#define QUEUE_SANITY_LIMIT 10
lws_sequencer_t *
lws_sequencer_create(lws_seq_info_t *i)
lws_seq_t *
lws_seq_create(lws_seq_info_t *i)
{
struct lws_context_per_thread *pt = &i->context->pt[i->tsi];
lws_sequencer_t *seq = lws_zalloc(sizeof(*seq) + i->user_size, __func__);
lws_seq_t *seq = lws_zalloc(sizeof(*seq) + i->user_size, __func__);
if (!seq)
return NULL;
@ -78,11 +78,11 @@ lws_sequencer_create(lws_seq_info_t *i)
lws_pt_unlock(pt); /* } pt ------------------------------------------ */
time(&seq->time_created);
seq->time_created = lws_now_usecs();
/* try to queue the creation cb */
if (lws_sequencer_queue_event(seq, LWSSEQ_CREATED, NULL, NULL)) {
if (lws_seq_queue_event(seq, LWSSEQ_CREATED, NULL, NULL)) {
lws_dll2_remove(&seq->seq_list);
lws_free(seq);
@ -105,9 +105,9 @@ seq_ev_destroy(struct lws_dll2 *d, void *user)
}
void
lws_sequencer_destroy(lws_sequencer_t **pseq)
lws_seq_destroy(lws_seq_t **pseq)
{
lws_sequencer_t *seq = *pseq;
lws_seq_t *seq = *pseq;
/* defeat another thread racing to add events while we are destroying */
seq->going_down = 1;
@ -129,20 +129,20 @@ lws_sequencer_destroy(lws_sequencer_t **pseq)
}
void
lws_sequencer_destroy_all_on_pt(struct lws_context_per_thread *pt)
lws_seq_destroy_all_on_pt(struct lws_context_per_thread *pt)
{
lws_start_foreach_dll_safe(struct lws_dll2 *, p, tp,
pt->seq_owner.head) {
lws_sequencer_t *s = lws_container_of(p, lws_sequencer_t,
lws_seq_t *s = lws_container_of(p, lws_seq_t,
seq_list);
lws_sequencer_destroy(&s);
lws_seq_destroy(&s);
} lws_end_foreach_dll_safe(p, tp);
}
int
lws_sequencer_queue_event(lws_sequencer_t *seq, lws_seq_events_t e, void *data,
lws_seq_queue_event(lws_seq_t *seq, lws_seq_events_t e, void *data,
void *aux)
{
lws_seq_event_t *seqe;
@ -191,7 +191,7 @@ lws_sequencer_queue_event(lws_sequencer_t *seq, lws_seq_events_t e, void *data,
*/
int
lws_sequencer_check_wsi(lws_sequencer_t *seq, struct lws *wsi)
lws_seq_check_wsi(lws_seq_t *seq, struct lws *wsi)
{
lws_seq_event_t *seqe;
struct lws_dll2 *dh;
@ -220,9 +220,9 @@ lws_sequencer_check_wsi(lws_sequencer_t *seq, struct lws *wsi)
*/
static int
lws_sequencer_next_event(struct lws_dll2 *d, void *user)
lws_seq_next_event(struct lws_dll2 *d, void *user)
{
lws_sequencer_t *seq = lws_container_of(d, lws_sequencer_t,
lws_seq_t *seq = lws_container_of(d, lws_seq_t,
seq_pend_list);
lws_seq_event_t *seqe;
struct lws_dll2 *dh;
@ -258,7 +258,7 @@ lws_sequencer_next_event(struct lws_dll2 *d, void *user)
if (n) {
lwsl_info("%s: destroying seq '%s' by request\n", __func__,
seq->name);
lws_sequencer_destroy(&seq);
lws_seq_destroy(&seq);
return LWSSEQ_RET_DESTROY;
}
@ -278,25 +278,24 @@ lws_pt_do_pending_sequencer_events(struct lws_context_per_thread *pt)
return 0;
return lws_dll2_foreach_safe(&pt->seq_pend_owner, NULL,
lws_sequencer_next_event);
lws_seq_next_event);
}
/* set secs to zero to remove timeout */
int
lws_sequencer_timeout(lws_sequencer_t *seq, int secs)
lws_seq_timeout_us(lws_seq_t *seq, lws_usec_t us)
{
lws_dll2_remove(&seq->seq_to_list);
if (!secs) {
if (!us) {
/* we are clearing the timeout */
seq->timeout = 0;
return 0;
}
time(&seq->timeout);
seq->timeout += secs;
seq->timeout = lws_now_usecs() + us;
/*
* we sort the pt's list of sequencers with pending timeouts, so it's
@ -305,14 +304,12 @@ lws_sequencer_timeout(lws_sequencer_t *seq, int secs)
lws_start_foreach_dll_safe(struct lws_dll2 *, p, tp,
seq->pt->seq_to_owner.head) {
lws_sequencer_t *s = lws_container_of(p, lws_sequencer_t,
seq_to_list);
lws_seq_t *s = lws_container_of(p, lws_seq_t, seq_to_list);
assert(s->timeout); /* shouldn't be on the list otherwise */
if (s->timeout >= seq->timeout) {
/* drop us in before this guy */
lws_dll2_add_before(&seq->seq_to_list,
&s->seq_to_list);
lws_dll2_add_before(&seq->seq_to_list, &s->seq_to_list);
return 0;
}
@ -330,61 +327,70 @@ lws_sequencer_timeout(lws_sequencer_t *seq, int secs)
/*
* nonpublic helper to check for and handle sequencer timeouts for a whole pt
* returns either 0 or number of us until next event (which cannot be 0 or we
* would have serviced it)
*/
int
lws_sequencer_timeout_check(struct lws_context_per_thread *pt, time_t now)
lws_usec_t
__lws_seq_timeout_check(struct lws_context_per_thread *pt, lws_usec_t usnow)
{
lws_usec_t future_us = 0;
lws_start_foreach_dll_safe(struct lws_dll2 *, p, tp,
pt->seq_to_owner.head) {
lws_sequencer_t *s = lws_container_of(p, lws_sequencer_t,
seq_to_list);
lws_seq_t *s = lws_container_of(p, lws_seq_t, seq_to_list);
assert(s->timeout); /* shouldn't be on the list otherwise */
if (s->timeout <= now) {
if (s->timeout <= usnow) {
/* seq has timed out... remove him from timeout list */
lws_sequencer_timeout(s, 0);
lws_seq_timeout_us(s, LWSSEQTO_NONE);
/* queue the message to inform the sequencer */
lws_sequencer_queue_event(s, LWSSEQ_TIMED_OUT,
NULL, NULL);
} else
lws_seq_queue_event(s, LWSSEQ_TIMED_OUT, NULL, NULL);
} else {
/*
* No need to look further if we met one later than now:
* the list is sorted in ascending time order
*/
return 0;
future_us = usnow - s->timeout;
break;
}
} lws_end_foreach_dll_safe(p, tp);
if (usnow - pt->last_heartbeat< LWS_US_PER_SEC)
return future_us;
pt->last_heartbeat = usnow;
/* send every sequencer a heartbeat message... it can ignore it */
lws_start_foreach_dll_safe(struct lws_dll2 *, p, tp,
pt->seq_owner.head) {
lws_sequencer_t *s = lws_container_of(p, lws_sequencer_t,
seq_list);
lws_seq_t *s = lws_container_of(p, lws_seq_t, seq_list);
/* queue the message to inform the sequencer */
lws_sequencer_queue_event(s, LWSSEQ_HEARTBEAT, NULL, NULL);
lws_seq_queue_event(s, LWSSEQ_HEARTBEAT, NULL, NULL);
} lws_end_foreach_dll_safe(p, tp);
return 0;
return future_us;
}
lws_sequencer_t *
lws_sequencer_from_user(void *u)
lws_seq_t *
lws_seq_from_user(void *u)
{
return &((lws_sequencer_t *)u)[-1];
return &((lws_seq_t *)u)[-1];
}
const char *
lws_sequencer_name(lws_sequencer_t *seq)
lws_seq_name(lws_seq_t *seq)
{
return seq->name;
}
int
lws_sequencer_secs_since_creation(lws_sequencer_t *seq)
lws_seq_secs_since_creation(lws_seq_t *seq)
{
time_t now;
@ -394,7 +400,7 @@ lws_sequencer_secs_since_creation(lws_sequencer_t *seq)
}
struct lws_context *
lws_sequencer_get_context(lws_sequencer_t *seq)
lws_seq_get_context(lws_seq_t *seq)
{
return seq->pt->context;
}

10
lib/core-net/service.c
View file

@ -603,6 +603,7 @@ lws_service_periodic_checks(struct lws_context *context,
lws_sockfd_type our_fd = 0, tmp_fd;
struct lws *wsi;
int timed_out = 0;
lws_usec_t usnow;
time_t now;
#if defined(LWS_ROLE_H1) || defined(LWS_ROLE_H2)
struct allocated_headers *ah;
@ -615,14 +616,15 @@ lws_service_periodic_checks(struct lws_context *context,
return -1;
}
time(&now);
usnow = lws_now_usecs();
now = usnow / LWS_US_PER_SEC;
/*
* handle case that system time was uninitialized when lws started
* at boot, and got initialized a little later
*/
if (context->time_up < 1464083026 && now > 1464083026)
context->time_up = now;
context->time_up = now / LWS_US_PER_SEC;
if (context->last_timeout_check_s &&
now - context->last_timeout_check_s > 100) {
@ -648,7 +650,7 @@ lws_service_periodic_checks(struct lws_context *context,
context->last_timeout_check_s = now - 1;
}
lws_sequencer_timeout_check(pt, now);
__lws_seq_timeout_check(pt, usnow);
lws_pt_do_pending_sequencer_events(pt);
if (!lws_compare_time_t(context, context->last_timeout_check_s, now))
@ -1086,7 +1088,7 @@ handled:
lws_service_periodic_checks(context, pollfd, tsi);
lws_pt_lock(pt, __func__);
__lws_hrtimer_service(pt);
__lws_hrtimer_service(pt, lws_now_usecs());
lws_pt_unlock(pt);
return 0;

20
lib/core-net/wsi-timeout.c
View file

@ -82,16 +82,12 @@ lws_set_timer_usecs(struct lws *wsi, lws_usec_t usecs)
__lws_set_timer_usecs(wsi, usecs);
}
/* return 0 if nothing pending, or the number of us before the next event */
lws_usec_t
__lws_hrtimer_service(struct lws_context_per_thread *pt)
__lws_hrtimer_service(struct lws_context_per_thread *pt, lws_usec_t t)
{
struct timeval now;
struct lws *wsi;
lws_usec_t t;
gettimeofday(&now, NULL);
t = (now.tv_sec * 1000000ll) + now.tv_usec;
lws_start_foreach_dll_safe(struct lws_dll *, d, d1,
pt->dll_hrtimer_head.next) {
@ -118,18 +114,16 @@ __lws_hrtimer_service(struct lws_context_per_thread *pt)
/* return an estimate how many us until next timer hit */
if (!pt->dll_hrtimer_head.next)
return LWS_HRTIMER_NOWAIT;
return 0; /* there is nothing pending */
wsi = lws_container_of(pt->dll_hrtimer_head.next, struct lws,
dll_hrtimer);
gettimeofday(&now, NULL);
t = (now.tv_sec * 1000000ll) + now.tv_usec;
t = lws_now_usecs();
if (wsi->pending_timer <= t) /* in the past */
return 1;
if (wsi->pending_timer < t)
return 0;
return wsi->pending_timer - t;
return wsi->pending_timer - t; /* at least 1 */
}
void

2
lib/core/context.c
View file

@ -552,7 +552,7 @@ lws_context_destroy3(struct lws_context *context)
for (n = 0; n < context->count_threads; n++) {
struct lws_context_per_thread *pt = &context->pt[n];
lws_sequencer_destroy_all_on_pt(pt);
lws_seq_destroy_all_on_pt(pt);
if (context->event_loop_ops->destroy_pt)
context->event_loop_ops->destroy_pt(context, n);

2
lib/core/private.h
View file

@ -241,8 +241,6 @@ lws_mutex_refcount_unlock(struct lws_mutex_refcount *mr);
#include "core-net/private.h"
#endif
#define LWS_HRTIMER_NOWAIT (0x7fffffffffffffffll)
struct lws_deferred_free
{
struct lws_deferred_free *next;

43
lib/event-libs/common.c Normal file
View file

@ -0,0 +1,43 @@
/*
* libwebsockets - small server side websockets and web server implementation
*
* Copyright (C) 2010-2018 Andy Green <andy@warmcat.com>
*
* This library is free software; you can redistribute it and/or
* modify it under the terms of the GNU Lesser General Public
* License as published by the Free Software Foundation:
* version 2.1 of the License.
*
* This library is distributed in the hope that it will be useful,
* but WITHOUT ANY WARRANTY; without even the implied warranty of
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
* Lesser General Public License for more details.
*
* You should have received a copy of the GNU Lesser General Public
* License along with this library; if not, write to the Free Software
* Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston,
* MA 02110-1301 USA
*/
#include "core/private.h"
lws_usec_t
__lws_event_service_get_earliest_wake(struct lws_context_per_thread *pt,
lws_usec_t usnow)
{
lws_usec_t t, us = 0;
char seen = 0;
t = __lws_hrtimer_service(pt, usnow);
if (t && (!seen || t < us)) {
us = t;
seen = 1;
}
t = __lws_seq_timeout_check(pt, usnow);
if (t && (!seen || t < us)) {
us = t;
seen = 1;
}
return us;
}

8
lib/event-libs/libev/libev.c
View file

@ -29,8 +29,8 @@ lws_ev_hrtimer_cb(struct ev_loop *loop, struct ev_timer *watcher, int revents)
lws_usec_t us;
lws_pt_lock(pt, __func__);
us = __lws_hrtimer_service(pt);
if (us != LWS_HRTIMER_NOWAIT) {
us = __lws_event_service_get_earliest_wake(pt, lws_now_usecs());
if (us) {
ev_timer_set(&pt->ev.hrtimer, ((float)us) / 1000000.0, 0);
ev_timer_start(pt->ev.io_loop, &pt->ev.hrtimer);
}
@ -61,8 +61,8 @@ lws_ev_idle_cb(struct ev_loop *loop, struct ev_idle *handle, int revents)
/* account for hrtimer */
lws_pt_lock(pt, __func__);
us = __lws_hrtimer_service(pt);
if (us != LWS_HRTIMER_NOWAIT) {
us = __lws_event_service_get_earliest_wake(pt, lws_now_usecs());
if (us) {
ev_timer_set(&pt->ev.hrtimer, ((float)us) / 1000000.0, 0);
ev_timer_start(pt->ev.io_loop, &pt->ev.hrtimer);
}

16
lib/event-libs/libevent/libevent.c
View file

@ -29,10 +29,10 @@ lws_event_hrtimer_cb(int fd, short event, void *p)
lws_usec_t us;
lws_pt_lock(pt, __func__);
us = __lws_hrtimer_service(pt);
if (us != LWS_HRTIMER_NOWAIT) {
tv.tv_sec = us / 1000000;
tv.tv_usec = us - (tv.tv_sec * 1000000);
us = __lws_event_service_get_earliest_wake(pt, lws_now_usecs());
if (us) {
tv.tv_sec = us / LWS_US_PER_SEC;
tv.tv_usec = us - (tv.tv_sec * LWS_US_PER_SEC);
evtimer_add(pt->event.hrtimer, &tv);
}
lws_pt_unlock(pt);
@ -70,10 +70,10 @@ lws_event_idle_timer_cb(int fd, short event, void *p)
/* account for hrtimer */
lws_pt_lock(pt, __func__);
us = __lws_hrtimer_service(pt);
if (us != LWS_HRTIMER_NOWAIT) {
tv.tv_sec = us / 1000000;
tv.tv_usec = us - (tv.tv_sec * 1000000);
us = __lws_event_service_get_earliest_wake(pt, lws_now_usecs());
if (us) {
tv.tv_sec = us / LWS_US_PER_SEC;
tv.tv_usec = us - (tv.tv_sec * LWS_US_PER_SEC);
evtimer_add(pt->event.hrtimer, &tv);
}
lws_pt_unlock(pt);

14
lib/event-libs/libuv/libuv.c
View file

@ -33,9 +33,10 @@ lws_uv_hrtimer_cb(uv_timer_t *timer
lws_usec_t us;
lws_pt_lock(pt, __func__);
us = __lws_hrtimer_service(pt);
if (us != LWS_HRTIMER_NOWAIT)
uv_timer_start(&pt->uv.hrtimer, lws_uv_hrtimer_cb, us / 1000, 0);
us = __lws_event_service_get_earliest_wake(pt, lws_now_usecs());
if (us)
uv_timer_start(&pt->uv.hrtimer, lws_uv_hrtimer_cb,
LWS_US_TO_MS(us), 0);
lws_pt_unlock(pt);
}
@ -67,9 +68,10 @@ lws_uv_idle(uv_idle_t *handle
/* account for hrtimer */
lws_pt_lock(pt, __func__);
us = __lws_hrtimer_service(pt);
if (us != LWS_HRTIMER_NOWAIT)
uv_timer_start(&pt->uv.hrtimer, lws_uv_hrtimer_cb, us / 1000, 0);
us = __lws_event_service_get_earliest_wake(pt, lws_now_usecs());
if (us)
uv_timer_start(&pt->uv.hrtimer, lws_uv_hrtimer_cb,
LWS_US_TO_MS(us), 0);
lws_pt_unlock(pt);
/* there is nobody who needs service forcing, shut down idle */

6
lib/event-libs/private.h
View file

@ -54,6 +54,12 @@ struct lws_event_loop_ops {
unsigned int periodic_events_available:1;
};
struct lws_context_per_thread;
lws_usec_t
__lws_event_service_get_earliest_wake(struct lws_context_per_thread *pt,
lws_usec_t usnow);
/* bring in event libs private declarations */
#if defined(LWS_WITH_POLL)

23
lib/plat/esp32/esp32-service.c
View file

@ -38,6 +38,7 @@ lws_plat_service(struct lws_context *context, int timeout_ms)
LWS_EXTERN int
_lws_plat_service_tsi(struct lws_context *context, int timeout_ms, int tsi)
{
lws_usec_t timeout_us = timeout_ms * LWS_US_PER_MS;
struct lws_context_per_thread *pt;
int n = -1, m, c;
@ -101,24 +102,28 @@ _lws_plat_service_tsi(struct lws_context *context, int timeout_ms, int tsi)
/* still somebody left who wants forced service? */
if (!lws_service_adjust_timeout(context, 1, pt->tid))
/* yes... come back again quickly */
timeout_ms = 0;
timeout_us = 0;
}
if (timeout_ms) {
if (timeout_us) {
lws_usec_t t, us = lws_now_usecs();
lws_pt_lock(pt, __func__);
/* don't stay in poll wait longer than next hr timeout */
lws_usec_t t = __lws_hrtimer_service(pt);
if ((lws_usec_t)timeout_ms * 1000 > t)
timeout_ms = t / 1000;
/* don't stay in poll wait longer than next hr timeout... */
t = __lws_hrtimer_service(pt, us);
if (t && timeout_us > t)
timeout_us = t;
/* ... or next sequencer timeout */
t = __lws_seq_timeout_check(pt, us);
if (t && timeout_us > t)
timeout_us = t;
lws_pt_unlock(pt);
}
// n = poll(pt->fds, pt->fds_count, timeout_ms);
{
fd_set readfds, writefds, errfds;
struct timeval tv = { timeout_ms / 1000,
(timeout_ms % 1000) * 1000 }, *ptv = &tv;
struct timeval tv = { timeout_us / LWS_US_PER_SEC,
timeout_us % LWS_US_PER_SEC }, *ptv = &tv;
int max_fd = 0;
FD_ZERO(&readfds);
FD_ZERO(&writefds);

19
lib/plat/optee/network.c
View file

@ -58,6 +58,7 @@ lws_poll_listen_fd(struct lws_pollfd *fd)
LWS_EXTERN int
_lws_plat_service_tsi(struct lws_context *context, int timeout_ms, int tsi)
{
lws_usec_t timeout_us = timeout_ms * LWS_US_PER_MS;
struct lws_context_per_thread *pt;
int n = -1, m, c;
//char buf;
@ -93,10 +94,24 @@ _lws_plat_service_tsi(struct lws_context *context, int timeout_ms, int tsi)
/* still somebody left who wants forced service? */
if (!lws_service_adjust_timeout(context, 1, pt->tid))
/* yes... come back again quickly */
timeout_ms = 0;
timeout_us = 0;
}
n = poll(pt->fds, pt->fds_count, timeout_ms);
if (timeout_us) {
lws_usec_t t, us = lws_now_usecs();
lws_pt_lock(pt, __func__);
/* don't stay in poll wait longer than next hr timeout */
t = __lws_hrtimer_service(pt, us);
if (t && timeout_us > t)
timeout_us = t;
t = __lws_seq_timeout_check(pt, us);
if (t && timeout_us > t)
timeout_us = t;
lws_pt_unlock(pt);
}
n = poll(pt->fds, pt->fds_count, timeout_us / LWS_US_PER_MS);
m = 0;

24
lib/plat/unix/unix-service.c
View file

@ -36,6 +36,7 @@ LWS_EXTERN int
_lws_plat_service_tsi(struct lws_context *context, int timeout_ms, int tsi)
{
volatile struct lws_foreign_thread_pollfd *ftp, *next;
lws_usec_t timeout_us = timeout_ms * LWS_US_PER_MS;
volatile struct lws_context_per_thread *vpt;
struct lws_context_per_thread *pt;
int n = -1, m, c;
@ -62,9 +63,9 @@ _lws_plat_service_tsi(struct lws_context *context, int timeout_ms, int tsi)
memset(&_lws, 0, sizeof(_lws));
_lws.context = context;
pt->service_tid =
context->vhost_list->protocols[0].callback(
&_lws, LWS_CALLBACK_GET_THREAD_ID, NULL, NULL, 0);
pt->service_tid = context->vhost_list->protocols[0].callback(
&_lws, LWS_CALLBACK_GET_THREAD_ID,
NULL, NULL, 0);
pt->service_tid_detected = 1;
}
@ -77,21 +78,24 @@ _lws_plat_service_tsi(struct lws_context *context, int timeout_ms, int tsi)
/* still somebody left who wants forced service? */
if (!lws_service_adjust_timeout(context, 1, pt->tid))
/* yes... come back again quickly */
timeout_ms = 0;
timeout_us = 0;
}
if (timeout_ms) {
if (timeout_us) {
lws_usec_t t;
lws_pt_lock(pt, __func__);
/* don't stay in poll wait longer than next hr timeout */
lws_usec_t t = __lws_hrtimer_service(pt);
if ((lws_usec_t)timeout_ms * 1000 > t)
timeout_ms = t / 1000;
t = __lws_event_service_get_earliest_wake(pt, lws_now_usecs());
if (t && t < timeout_us)
timeout_us = t;
lws_pt_unlock(pt);
}
vpt->inside_poll = 1;
lws_memory_barrier();
n = poll(pt->fds, pt->fds_count, timeout_ms);
n = poll(pt->fds, pt->fds_count, timeout_us / LWS_US_PER_MS);
vpt->inside_poll = 0;
lws_memory_barrier();
@ -130,7 +134,7 @@ _lws_plat_service_tsi(struct lws_context *context, int timeout_ms, int tsi)
/* we have come out of a poll wait... check the hrtimer list */
__lws_hrtimer_service(pt);
__lws_hrtimer_service(pt, lws_now_usecs());
lws_pt_unlock(pt);

23
lib/plat/windows/windows-service.c
View file

@ -28,6 +28,7 @@
LWS_EXTERN int
_lws_plat_service_tsi(struct lws_context *context, int timeout_ms, int tsi)
{
lws_usec_t timeout_us = timeout_ms * LWS_US_PER_MS;
struct lws_context_per_thread *pt;
WSANETWORKEVENTS networkevents;
struct lws_pollfd *pfd;
@ -114,18 +115,21 @@ _lws_plat_service_tsi(struct lws_context *context, int timeout_ms, int tsi)
/* still somebody left who wants forced service? */
if (!lws_service_adjust_timeout(context, 1, pt->tid))
/* yes... come back again quickly */
timeout_ms = 0;
timeout_us = 0;
}
if (timeout_ms) {
lws_usec_t t;
if (timeout_us) {
lws_usec_t t, us = lws_now_usecs();
lws_pt_lock(pt, __func__);
/* don't stay in poll wait longer than next hr timeout */
t = __lws_hrtimer_service(pt);
if ((lws_usec_t)timeout_ms * 1000 > t)
timeout_ms = (int)(t / 1000);
/* don't stay in poll wait longer than next hr timeout... */
t = __lws_hrtimer_service(pt, us);
if (t && timeout_us > t)
timeout_us = t;
/* ... or next sequencer timeout */
t = __lws_seq_timeout_check(pt, us);
if (t && timeout_us > t)
timeout_us = t;
lws_pt_unlock(pt);
}
@ -137,7 +141,8 @@ _lws_plat_service_tsi(struct lws_context *context, int timeout_ms, int tsi)
FD_ROUTING_INTERFACE_CHANGE |
FD_ADDRESS_LIST_CHANGE);
ev = WSAWaitForMultipleEvents(1, &pt->events, FALSE, timeout_ms, FALSE);
ev = WSAWaitForMultipleEvents(1, &pt->events, FALSE,
timeout_us / LWS_US_PER_MS, FALSE);
if (ev == WSA_WAIT_EVENT_0) {
EnterCriticalSection(&pt->interrupt_lock);
interrupt_requested = pt->interrupt_requested;

14
minimal-examples/api-tests/api-test-lws_sequencer/main.c
View file

@ -162,7 +162,7 @@ notify:
lws_set_wsi_user(wsi, NULL);
s->cwsi = NULL;
lws_sequencer_queue_event(lws_sequencer_from_user(s), seq_msg,
lws_seq_queue_event(lws_seq_from_user(s), seq_msg,
NULL, NULL);
return 0;
@ -185,7 +185,7 @@ sequencer_start_client(struct myseq *s)
lws_strncpy(uri, url_paths[s->state], sizeof(uri));
memset(&i, 0, sizeof i);
i.context = lws_sequencer_get_context(lws_sequencer_from_user(s));
i.context = lws_seq_get_context(lws_seq_from_user(s));
if (lws_parse_uri(uri, &prot, &i.address, &i.port, &path1)) {
lwsl_err("%s: uri error %s\n", __func__, uri);
@ -217,13 +217,13 @@ sequencer_start_client(struct myseq *s)
/* we couldn't even get started with the client connection */
lws_sequencer_queue_event(lws_sequencer_from_user(s),
lws_seq_queue_event(lws_seq_from_user(s),
SEQ_MSG_CLIENT_FAILED, NULL, NULL);
return 1;
}
lws_sequencer_timeout(lws_sequencer_from_user(s), 3);
lws_seq_timeout_us(lws_seq_from_user(s), 3 * LWS_US_PER_SEC);
lwsl_notice("%s: wsi %p: connecting to %s://%s:%d%s\n", __func__,
s->cwsi, prot, i.address, i.port, path);
@ -297,7 +297,7 @@ sequencer_cb(struct lws_sequencer *seq, void *user, int event,
s->state, s->http_resp);
done:
lws_sequencer_timeout(lws_sequencer_from_user(s), 0);
lws_seq_timeout_us(lws_seq_from_user(s), LWSSEQTO_NONE);
s->state++;
if (s->state == LWS_ARRAY_SIZE(url_paths)) {
/* the sequence has completed */
@ -324,7 +324,7 @@ main(int argc, const char **argv)
int n = 1, logs = LLL_USER | LLL_ERR | LLL_WARN | LLL_NOTICE;
struct lws_context_creation_info info;
struct lws_context *context;
lws_sequencer_t *seq;
lws_seq_t *seq;
struct lws_vhost *vh;
lws_seq_info_t i;
struct myseq *s;
@ -378,7 +378,7 @@ main(int argc, const char **argv)
i.cb = sequencer_cb;
i.name = "seq";
seq = lws_sequencer_create(&i);
seq = lws_seq_create(&i);
if (!seq) {
lwsl_err("%s: unable to create sequencer\n", __func__);
goto bail1;