lib/core-net/sequencer.c

   1 /*
   2  * libwebsockets - lib/core-net/sequencer.c
   3  *
   4  * Copyright (C) 2019 Andy Green <andy@warmcat.com>
   5  *
   6  *  This library is free software; you can redistribute it and/or
   7  *  modify it under the terms of the GNU Lesser General Public
   8  *  License as published by the Free Software Foundation:
   9  *  version 2.1 of the License.
  10  *
  11  *  This library is distributed in the hope that it will be useful,
  12  *  but WITHOUT ANY WARRANTY; without even the implied warranty of
  13  *  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
  14  *  Lesser General Public License for more details.
  15  *
  16  *  You should have received a copy of the GNU Lesser General Public
  17  *  License along with this library; if not, write to the Free Software
  18  *  Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston,
  19  *  MA  02110-1301  USA
  20  */
  21
  22 #include "private-lib-core.h"
  23
  24 /*
  25  * per pending event
  26  */
  27 typedef struct lws_seq_event {
  28         struct lws_dll2                 seq_event_list;
  29
  30         void                            *data;
  31         void                            *aux;
  32         lws_seq_events_t                e;
  33 } lws_seq_event_t;
  34
  35 /*
  36  * per sequencer
  37  */
  38 typedef struct lws_sequencer {
  39         struct lws_dll2                 seq_list;
  40
  41         lws_sorted_usec_list_t          sul_timeout;
  42         lws_sorted_usec_list_t          sul_pending;
  43
  44         struct lws_dll2_owner           seq_event_owner;
  45         struct lws_context_per_thread   *pt;
  46         lws_seq_event_cb                cb;
  47         const char                      *name;
  48         const lws_retry_bo_t            *retry;
  49
  50         lws_usec_t                      time_created;
  51         lws_usec_t                      timeout; /* 0 or time we timeout */
  52
  53         char                            going_down;
  54 } lws_seq_t;
  55
  56 #define QUEUE_SANITY_LIMIT 10
  57
  58 static void
  59 lws_sul_seq_heartbeat_cb(lws_sorted_usec_list_t *sul)
  60 {
  61         struct lws_context_per_thread *pt = lws_container_of(sul,
  62                         struct lws_context_per_thread, sul_seq_heartbeat);
  63
  64         /* send every sequencer a heartbeat message... it can ignore it */
  65
  66         lws_start_foreach_dll_safe(struct lws_dll2 *, p, tp,
  67                                    lws_dll2_get_head(&pt->seq_owner)) {
  68                 lws_seq_t *s = lws_container_of(p, lws_seq_t, seq_list);
  69
  70                 /* queue the message to inform the sequencer */
  71                 lws_seq_queue_event(s, LWSSEQ_HEARTBEAT, NULL, NULL);
  72
  73         } lws_end_foreach_dll_safe(p, tp);
  74
  75         /* schedule the next one */
  76
  77         __lws_sul_insert(&pt->pt_sul_owner, &pt->sul_seq_heartbeat,
  78                          LWS_US_PER_SEC);
  79 }
  80
  81 int
  82 lws_seq_pt_init(struct lws_context_per_thread *pt)
  83 {
  84         pt->sul_seq_heartbeat.cb = lws_sul_seq_heartbeat_cb;
  85
  86         /* schedule the first heartbeat */
  87         __lws_sul_insert(&pt->pt_sul_owner, &pt->sul_seq_heartbeat,
  88                          LWS_US_PER_SEC);
  89
  90         return 0;
  91 }
  92
  93 lws_seq_t *
  94 lws_seq_create(lws_seq_info_t *i)
  95 {
  96         struct lws_context_per_thread *pt = &i->context->pt[i->tsi];
  97         lws_seq_t *seq = lws_zalloc(sizeof(*seq) + i->user_size, __func__);
  98
  99         if (!seq)
 100                 return NULL;
 101
 102         seq->cb = i->cb;
 103         seq->pt = pt;
 104         seq->name = i->name;
 105         seq->retry = i->retry;
 106
 107         *i->puser = (void *)&seq[1];
 108
 109         /* add the sequencer to the pt */
 110
 111         lws_pt_lock(pt, __func__); /* ---------------------------------- pt { */
 112
 113         lws_dll2_add_tail(&seq->seq_list, &pt->seq_owner);
 114
 115         lws_pt_unlock(pt); /* } pt ------------------------------------------ */
 116
 117         seq->time_created = lws_now_usecs();
 118
 119         /* try to queue the creation cb */
 120
 121         if (lws_seq_queue_event(seq, LWSSEQ_CREATED, NULL, NULL)) {
 122                 lws_dll2_remove(&seq->seq_list);
 123                 lws_free(seq);
 124
 125                 return NULL;
 126         }
 127
 128         return seq;
 129 }
 130
 131 static int
 132 seq_ev_destroy(struct lws_dll2 *d, void *user)
 133 {
 134         lws_seq_event_t *seqe = lws_container_of(d, lws_seq_event_t,
 135                                                  seq_event_list);
 136
 137         lws_dll2_remove(&seqe->seq_event_list);
 138         lws_free(seqe);
 139
 140         return 0;
 141 }
 142
 143 void
 144 lws_seq_destroy(lws_seq_t **pseq)
 145 {
 146         lws_seq_t *seq = *pseq;
 147
 148         /* defeat another thread racing to add events while we are destroying */
 149         seq->going_down = 1;
 150
 151         seq->cb(seq, (void *)&seq[1], LWSSEQ_DESTROYED, NULL, NULL);
 152
 153         lws_pt_lock(seq->pt, __func__); /* -------------------------- pt { */
 154
 155         lws_dll2_remove(&seq->seq_list);
 156         lws_dll2_remove(&seq->sul_timeout.list);
 157         lws_dll2_remove(&seq->sul_pending.list);
 158         /* remove and destroy any pending events */
 159         lws_dll2_foreach_safe(&seq->seq_event_owner, NULL, seq_ev_destroy);
 160
 161         lws_pt_unlock(seq->pt); /* } pt ---------------------------------- */
 162
 163
 164         lws_free_set_NULL(seq);
 165 }
 166
 167 void
 168 lws_seq_destroy_all_on_pt(struct lws_context_per_thread *pt)
 169 {
 170         lws_start_foreach_dll_safe(struct lws_dll2 *, p, tp,
 171                                    pt->seq_owner.head) {
 172                 lws_seq_t *s = lws_container_of(p, lws_seq_t,
 173                                                       seq_list);
 174
 175                 lws_seq_destroy(&s);
 176
 177         } lws_end_foreach_dll_safe(p, tp);
 178 }
 179
 180 static void
 181 lws_seq_sul_pending_cb(lws_sorted_usec_list_t *sul)
 182 {
 183         lws_seq_t *seq = lws_container_of(sul, lws_seq_t, sul_pending);
 184         lws_seq_event_t *seqe;
 185         struct lws_dll2 *dh;
 186         int n;
 187
 188         if (!seq->seq_event_owner.count)
 189                 return;
 190
 191         /* events are only added at tail, so no race possible yet... */
 192
 193         dh = lws_dll2_get_head(&seq->seq_event_owner);
 194         seqe = lws_container_of(dh, lws_seq_event_t, seq_event_list);
 195
 196         n = seq->cb(seq, (void *)&seq[1], seqe->e, seqe->data, seqe->aux);
 197
 198         /* ... have to lock here though, because we will change the list */
 199
 200         lws_pt_lock(seq->pt, __func__); /* ----------------------------- pt { */
 201
 202         /* detach event from sequencer event list and free it */
 203         lws_dll2_remove(&seqe->seq_event_list);
 204         lws_free(seqe);
 205         lws_pt_unlock(seq->pt); /* } pt ------------------------------------- */
 206
 207         if (n) {
 208                 lwsl_info("%s: destroying seq '%s' by request\n", __func__,
 209                                 seq->name);
 210                 lws_seq_destroy(&seq);
 211         }
 212 }
 213
 214 int
 215 lws_seq_queue_event(lws_seq_t *seq, lws_seq_events_t e, void *data, void *aux)
 216 {
 217         lws_seq_event_t *seqe;
 218
 219         if (!seq || seq->going_down)
 220                 return 1;
 221
 222         seqe = lws_zalloc(sizeof(*seqe), __func__);
 223         if (!seqe)
 224                 return 1;
 225
 226         seqe->e = e;
 227         seqe->data = data;
 228         seqe->aux = aux;
 229
 230         // lwsl_notice("%s: seq %s: event %d\n", __func__, seq->name, e);
 231
 232         lws_pt_lock(seq->pt, __func__); /* ----------------------------- pt { */
 233
 234         if (seq->seq_event_owner.count > QUEUE_SANITY_LIMIT) {
 235                 lwsl_err("%s: more than %d events queued\n", __func__,
 236                          QUEUE_SANITY_LIMIT);
 237         }
 238
 239         lws_dll2_add_tail(&seqe->seq_event_list, &seq->seq_event_owner);
 240
 241         seq->sul_pending.cb = lws_seq_sul_pending_cb;
 242         __lws_sul_insert(&seq->pt->pt_sul_owner, &seq->sul_pending, 1);
 243
 244         lws_pt_unlock(seq->pt); /* } pt ------------------------------------- */
 245
 246         return 0;
 247 }
 248
 249 /*
 250  * Check if wsi still extant, by peeking in the message queue for a
 251  * LWSSEQ_WSI_CONN_CLOSE message about wsi.  (Doesn't need to do the same for
 252  * CONN_FAIL since that will never have produced any messages prior to that).
 253  *
 254  * Use this to avoid trying to perform operations on wsi that have already
 255  * closed but we didn't get to that message yet.
 256  *
 257  * Returns 0 if not closed yet or 1 if it has closed but we didn't process the
 258  * close message yet.
 259  */
 260
 261 int
 262 lws_seq_check_wsi(lws_seq_t *seq, struct lws *wsi)
 263 {
 264         lws_seq_event_t *seqe;
 265         struct lws_dll2 *dh;
 266
 267         lws_pt_lock(seq->pt, __func__); /* ----------------------------- pt { */
 268
 269         dh = lws_dll2_get_head(&seq->seq_event_owner);
 270         while (dh) {
 271                 seqe = lws_container_of(dh, lws_seq_event_t, seq_event_list);
 272
 273                 if (seqe->e == LWSSEQ_WSI_CONN_CLOSE && seqe->data == wsi)
 274                         break;
 275
 276                 dh = dh->next;
 277         }
 278
 279         lws_pt_unlock(seq->pt); /* } pt ------------------------------------- */
 280
 281         return !!dh;
 282 }
 283
 284
 285 static void
 286 lws_seq_sul_timeout_cb(lws_sorted_usec_list_t *sul)
 287 {
 288         lws_seq_t *s = lws_container_of(sul, lws_seq_t, sul_timeout);
 289
 290         lws_seq_queue_event(s, LWSSEQ_TIMED_OUT, NULL, NULL);
 291 }
 292
 293 /* set us to LWS_SET_TIMER_USEC_CANCEL to remove timeout */
 294
 295 int
 296 lws_seq_timeout_us(lws_seq_t *seq, lws_usec_t us)
 297 {
 298         seq->sul_timeout.cb = lws_seq_sul_timeout_cb;
 299         /* list is always at the very top of the sul */
 300         return __lws_sul_insert(&seq->pt->pt_sul_owner,
 301                         (lws_sorted_usec_list_t *)&seq->sul_timeout.list, us);
 302 }
 303
 304 lws_seq_t *
 305 lws_seq_from_user(void *u)
 306 {
 307         return &((lws_seq_t *)u)[-1];
 308 }
 309
 310 const char *
 311 lws_seq_name(lws_seq_t *seq)
 312 {
 313         return seq->name;
 314 }
 315
 316 lws_usec_t
 317 lws_seq_us_since_creation(lws_seq_t *seq)
 318 {
 319         return lws_now_usecs() - seq->time_created;
 320 }
 321
 322 struct lws_context *
 323 lws_seq_get_context(lws_seq_t *seq)
 324 {
 325         return seq->pt->context;
 326 }
 327