lib/pollfd.c

   1 /*
   2  * libwebsockets - small server side websockets and web server implementation
   3  *
   4  * Copyright (C) 2010-2015 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-libwebsockets.h"
  23
  24 int
  25 _lws_change_pollfd(struct lws *wsi, int _and, int _or, struct lws_pollargs *pa)
  26 {
  27         struct lws_context_per_thread *pt;
  28         struct lws_context *context;
  29         int ret = 0, pa_events = 1;
  30         struct lws_pollfd *pfd;
  31         int sampled_tid, tid;
  32
  33         if (!wsi || wsi->position_in_fds_table < 0)
  34                 return 0;
  35
  36         context = wsi->context;
  37         pt = &context->pt[(int)wsi->tsi];
  38         assert(wsi->position_in_fds_table >= 0 &&
  39                wsi->position_in_fds_table < pt->fds_count);
  40
  41         pfd = &pt->fds[wsi->position_in_fds_table];
  42         pa->fd = wsi->sock;
  43         pa->prev_events = pfd->events;
  44         pa->events = pfd->events = (pfd->events & ~_and) | _or;
  45
  46
  47         if (wsi->http2_substream)
  48                 return 0;
  49
  50         if (wsi->vhost->protocols[0].callback(wsi, LWS_CALLBACK_CHANGE_MODE_POLL_FD,
  51                                            wsi->user_space, (void *)pa, 0)) {
  52                 ret = -1;
  53                 goto bail;
  54         }
  55
  56         if (_and & LWS_POLLIN) {
  57                 lws_libev_io(wsi, LWS_EV_STOP | LWS_EV_READ);
  58                 lws_libuv_io(wsi, LWS_EV_STOP | LWS_EV_READ);
  59         }
  60         if (_or & LWS_POLLIN) {
  61                 lws_libev_io(wsi, LWS_EV_START | LWS_EV_READ);
  62                 lws_libuv_io(wsi, LWS_EV_START | LWS_EV_READ);
  63         }
  64         if (_and & LWS_POLLOUT) {
  65                 lws_libev_io(wsi, LWS_EV_STOP | LWS_EV_WRITE);
  66                 lws_libuv_io(wsi, LWS_EV_STOP | LWS_EV_WRITE);
  67         }
  68         if (_or & LWS_POLLOUT) {
  69                 lws_libev_io(wsi, LWS_EV_START | LWS_EV_WRITE);
  70                 lws_libuv_io(wsi, LWS_EV_START | LWS_EV_WRITE);
  71         }
  72
  73         /*
  74          * if we changed something in this pollfd...
  75          *   ... and we're running in a different thread context
  76          *     than the service thread...
  77          *       ... and the service thread is waiting ...
  78          *         then cancel it to force a restart with our changed events
  79          */
  80 #if LWS_POSIX
  81         pa_events = pa->prev_events != pa->events;
  82 #endif
  83
  84         if (pa_events) {
  85
  86                 if (lws_plat_change_pollfd(context, wsi, pfd)) {
  87                         lwsl_info("%s failed\n", __func__);
  88                         ret = -1;
  89                         goto bail;
  90                 }
  91
  92                 sampled_tid = context->service_tid;
  93                 if (sampled_tid) {
  94                         tid = wsi->vhost->protocols[0].callback(wsi,
  95                                      LWS_CALLBACK_GET_THREAD_ID, NULL, NULL, 0);
  96                         if (tid == -1) {
  97                                 ret = -1;
  98                                 goto bail;
  99                         }
 100                         if (tid != sampled_tid)
 101                                 lws_cancel_service_pt(wsi);
 102                 }
 103         }
 104 bail:
 105         return ret;
 106 }
 107
 108 int
 109 insert_wsi_socket_into_fds(struct lws_context *context, struct lws *wsi)
 110 {
 111         struct lws_pollargs pa = { wsi->sock, LWS_POLLIN, 0 };
 112         struct lws_context_per_thread *pt = &context->pt[(int)wsi->tsi];
 113         int ret = 0;
 114 #ifndef LWS_NO_SERVER
 115         struct lws_pollargs pa1;
 116 #endif
 117
 118         lwsl_debug("%s: %p: tsi=%d, sock=%d, pos-in-fds=%d\n",
 119                   __func__, wsi, wsi->tsi, wsi->sock, pt->fds_count);
 120
 121         if ((unsigned int)pt->fds_count >= context->fd_limit_per_thread) {
 122                 lwsl_err("Too many fds (%d vs %d)\n", context->max_fds,
 123                                 context->fd_limit_per_thread    );
 124                 return 1;
 125         }
 126
 127 #if !defined(_WIN32) && !defined(MBED_OPERATORS)
 128         if (wsi->sock >= context->max_fds) {
 129                 lwsl_err("Socket fd %d is too high (%d)\n",
 130                          wsi->sock, context->max_fds);
 131                 return 1;
 132         }
 133 #endif
 134
 135         assert(wsi);
 136         assert(wsi->vhost);
 137         assert(lws_socket_is_valid(wsi->sock));
 138
 139         if (wsi->vhost->protocols[0].callback(wsi, LWS_CALLBACK_LOCK_POLL,
 140                                            wsi->user_space, (void *) &pa, 1))
 141                 return -1;
 142
 143         lws_pt_lock(pt);
 144         pt->count_conns++;
 145         insert_wsi(context, wsi);
 146         wsi->position_in_fds_table = pt->fds_count;
 147         pt->fds[pt->fds_count].fd = wsi->sock;
 148         pt->fds[pt->fds_count].events = LWS_POLLIN;
 149         pa.events = pt->fds[pt->fds_count].events;
 150
 151         lws_plat_insert_socket_into_fds(context, wsi);
 152
 153         /* external POLL support via protocol 0 */
 154         if (wsi->vhost->protocols[0].callback(wsi, LWS_CALLBACK_ADD_POLL_FD,
 155                                            wsi->user_space, (void *) &pa, 0))
 156                 ret =  -1;
 157 #ifndef LWS_NO_SERVER
 158         /* if no more room, defeat accepts on this thread */
 159         if ((unsigned int)pt->fds_count == context->fd_limit_per_thread - 1)
 160                 _lws_change_pollfd(pt->wsi_listening, LWS_POLLIN, 0, &pa1);
 161 #endif
 162         lws_pt_unlock(pt);
 163
 164         if (wsi->vhost->protocols[0].callback(wsi, LWS_CALLBACK_UNLOCK_POLL,
 165                                            wsi->user_space, (void *)&pa, 1))
 166                 ret = -1;
 167
 168         return ret;
 169 }
 170
 171 int
 172 remove_wsi_socket_from_fds(struct lws *wsi)
 173 {
 174         struct lws_pollargs pa = { wsi->sock, 0, 0 };
 175 #ifndef LWS_NO_SERVER
 176         struct lws_pollargs pa1;
 177 #endif
 178         struct lws_context *context = wsi->context;
 179         struct lws_context_per_thread *pt = &context->pt[(int)wsi->tsi];
 180         struct lws *end_wsi;
 181         int m, ret = 0;
 182
 183 #if !defined(_WIN32) && !defined(MBED_OPERATORS)
 184         if (wsi->sock > context->max_fds) {
 185                 lwsl_err("fd %d too high (%d)\n", wsi->sock, context->max_fds);
 186                 return 1;
 187         }
 188 #endif
 189
 190         if (wsi->vhost->protocols[0].callback(wsi, LWS_CALLBACK_LOCK_POLL,
 191                                            wsi->user_space, (void *)&pa, 1))
 192                 return -1;
 193
 194         /*
 195          * detach ourselves from vh protocol list if we're on one
 196          * A -> B -> C
 197          * A -> C , or, B -> C, or A -> B
 198          */
 199         lwsl_info("%s: removing same prot wsi %p\n", __func__, wsi);
 200         if (wsi->same_vh_protocol_prev) {
 201                 assert (*(wsi->same_vh_protocol_prev) == wsi);
 202                 lwsl_info("have prev %p, setting him to our next %p\n",
 203                          wsi->same_vh_protocol_prev,
 204                          wsi->same_vh_protocol_next);
 205
 206                 /* guy who pointed to us should point to our next */
 207                 *(wsi->same_vh_protocol_prev) = wsi->same_vh_protocol_next;
 208         } //else
 209                 //lwsl_err("null wsi->prev\n");
 210         /* our next should point back to our prev */
 211         if (wsi->same_vh_protocol_next) {
 212                 lwsl_info("have next %p\n");
 213                 wsi->same_vh_protocol_next->same_vh_protocol_prev =
 214                                 wsi->same_vh_protocol_prev;
 215         } //else
 216                 //lwsl_err("null wsi->next\n");
 217
 218         lws_libev_io(wsi, LWS_EV_STOP | LWS_EV_READ | LWS_EV_WRITE | LWS_EV_PREPARE_DELETION);
 219         lws_libuv_io(wsi, LWS_EV_STOP | LWS_EV_READ | LWS_EV_WRITE | LWS_EV_PREPARE_DELETION);
 220
 221         lws_pt_lock(pt);
 222
 223         lwsl_info("%s: wsi=%p, sock=%d, fds pos=%d, end guy pos=%d, endfd=%d\n",
 224                   __func__, wsi, wsi->sock, wsi->position_in_fds_table,
 225                   pt->fds_count, pt->fds[pt->fds_count].fd);
 226
 227         /* the guy who is to be deleted's slot index in pt->fds */
 228         m = wsi->position_in_fds_table;
 229
 230         /* have the last guy take up the now vacant slot */
 231         pt->fds[m] = pt->fds[pt->fds_count - 1];
 232
 233         lws_plat_delete_socket_from_fds(context, wsi, m);
 234
 235         /* end guy's "position in fds table" is now the deletion guy's old one */
 236         end_wsi = wsi_from_fd(context, pt->fds[pt->fds_count].fd);
 237         assert(end_wsi);
 238         end_wsi->position_in_fds_table = m;
 239
 240         /* deletion guy's lws_lookup entry needs nuking */
 241         delete_from_fd(context, wsi->sock);
 242         /* removed wsi has no position any more */
 243         wsi->position_in_fds_table = -1;
 244
 245         /* remove also from external POLL support via protocol 0 */
 246         if (lws_socket_is_valid(wsi->sock))
 247                 if (wsi->vhost->protocols[0].callback(wsi, LWS_CALLBACK_DEL_POLL_FD,
 248                                                    wsi->user_space, (void *) &pa, 0))
 249                         ret = -1;
 250 #ifndef LWS_NO_SERVER
 251         if (!context->being_destroyed)
 252                 /* if this made some room, accept connects on this thread */
 253                 if ((unsigned int)pt->fds_count < context->fd_limit_per_thread - 1)
 254                         _lws_change_pollfd(pt->wsi_listening, 0, LWS_POLLIN, &pa1);
 255 #endif
 256         lws_pt_unlock(pt);
 257
 258         if (wsi->vhost->protocols[0].callback(wsi, LWS_CALLBACK_UNLOCK_POLL,
 259                                            wsi->user_space, (void *) &pa, 1))
 260                 ret = -1;
 261
 262         return ret;
 263 }
 264
 265 int
 266 lws_change_pollfd(struct lws *wsi, int _and, int _or)
 267 {
 268         struct lws_context_per_thread *pt;
 269         struct lws_context *context;
 270         struct lws_pollargs pa;
 271         int ret = 0;
 272
 273         if (!wsi || !wsi->protocol || wsi->position_in_fds_table < 0)
 274                 return 1;
 275
 276         context = lws_get_context(wsi);
 277         if (!context)
 278                 return 1;
 279
 280         if (wsi->vhost->protocols[0].callback(wsi, LWS_CALLBACK_LOCK_POLL,
 281                                            wsi->user_space,  (void *) &pa, 0))
 282                 return -1;
 283
 284         pt = &context->pt[(int)wsi->tsi];
 285
 286         lws_pt_lock(pt);
 287         ret = _lws_change_pollfd(wsi, _and, _or, &pa);
 288         lws_pt_unlock(pt);
 289         if (wsi->vhost->protocols[0].callback(wsi, LWS_CALLBACK_UNLOCK_POLL,
 290                                            wsi->user_space, (void *) &pa, 0))
 291                 ret = -1;
 292
 293         return ret;
 294 }
 295
 296
 297 /**
 298  * lws_callback_on_writable() - Request a callback when this socket
 299  *                                       becomes able to be written to without
 300  *                                       blocking
 301  *
 302  * @wsi:        Websocket connection instance to get callback for
 303  */
 304
 305 LWS_VISIBLE int
 306 lws_callback_on_writable(struct lws *wsi)
 307 {
 308 #ifdef LWS_USE_HTTP2
 309         struct lws *network_wsi, *wsi2;
 310         int already;
 311 #endif
 312
 313         if (wsi->state == LWSS_SHUTDOWN)
 314                 return 0;
 315
 316         if (wsi->socket_is_permanently_unusable)
 317                 return 0;
 318
 319 #ifdef LWS_USE_HTTP2
 320         lwsl_info("%s: %p\n", __func__, wsi);
 321
 322         if (wsi->mode != LWSCM_HTTP2_SERVING)
 323                 goto network_sock;
 324
 325         if (wsi->u.http2.requested_POLLOUT) {
 326                 lwsl_info("already pending writable\n");
 327                 return 1;
 328         }
 329
 330         if (wsi->u.http2.tx_credit <= 0) {
 331                 /*
 332                  * other side is not able to cope with us sending
 333                  * anything so no matter if we have POLLOUT on our side.
 334                  *
 335                  * Delay waiting for our POLLOUT until peer indicates he has
 336                  * space for more using tx window command in http2 layer
 337                  */
 338                 lwsl_info("%s: %p: waiting_tx_credit (%d)\n", __func__, wsi,
 339                           wsi->u.http2.tx_credit);
 340                 wsi->u.http2.waiting_tx_credit = 1;
 341                 return 0;
 342         }
 343
 344         network_wsi = lws_http2_get_network_wsi(wsi);
 345         already = network_wsi->u.http2.requested_POLLOUT;
 346
 347         /* mark everybody above him as requesting pollout */
 348
 349         wsi2 = wsi;
 350         while (wsi2) {
 351                 wsi2->u.http2.requested_POLLOUT = 1;
 352                 lwsl_info("mark %p pending writable\n", wsi2);
 353                 wsi2 = wsi2->u.http2.parent_wsi;
 354         }
 355
 356         /* for network action, act only on the network wsi */
 357
 358         wsi = network_wsi;
 359         if (already)
 360                 return 1;
 361 network_sock:
 362 #endif
 363
 364         if (lws_ext_cb_active(wsi, LWS_EXT_CB_REQUEST_ON_WRITEABLE, NULL, 0))
 365                 return 1;
 366
 367         if (wsi->position_in_fds_table < 0) {
 368                 lwsl_err("%s: failed to find socket %d\n", __func__, wsi->sock);
 369                 return -1;
 370         }
 371
 372         if (lws_change_pollfd(wsi, 0, LWS_POLLOUT))
 373                 return -1;
 374
 375         return 1;
 376 }
 377
 378 /**
 379  * lws_callback_on_writable_all_protocol_vhost() - Request a callback for
 380  *                      all connections using the given protocol when it
 381  *                      becomes possible to write to each socket without
 382  *                      blocking in turn.
 383  *
 384  *      This calls back connections with the same protocol ON THE SAME
 385  *      VHOST ONLY.
 386  *
 387  * @vhost:      Only consider connections on this lws_vhost
 388  * @protocol:   Protocol whose connections will get callbacks
 389  */
 390
 391 LWS_VISIBLE int
 392 lws_callback_on_writable_all_protocol_vhost(const struct lws_vhost *vhost,
 393                                       const struct lws_protocols *protocol)
 394 {
 395         struct lws *wsi;
 396
 397         if (protocol < vhost->protocols ||
 398             protocol >= (vhost->protocols + vhost->count_protocols)) {
 399                 lwsl_err("%s: protocol is not from vhost\n", __func__);
 400
 401                 return -1;
 402         }
 403
 404         wsi = vhost->same_vh_protocol_list[protocol - vhost->protocols];
 405         //lwsl_notice("%s: protocol %p, start wsi %p\n", __func__, protocol, wsi);
 406         while (wsi) {
 407                 //lwsl_notice("%s: protocol %p, this wsi %p (wsi->protocol=%p)\n",
 408                 //              __func__, protocol, wsi, wsi->protocol);
 409                 assert(wsi->protocol == protocol);
 410                 assert(*wsi->same_vh_protocol_prev == wsi);
 411                 if (wsi->same_vh_protocol_next) {
 412                 //      lwsl_err("my next says %p\n", wsi->same_vh_protocol_next);
 413                 //      lwsl_err("my next's prev says %p\n",
 414                 //              wsi->same_vh_protocol_next->same_vh_protocol_prev);
 415                         assert(wsi->same_vh_protocol_next->same_vh_protocol_prev == &wsi->same_vh_protocol_next);
 416                 }
 417                 lws_callback_on_writable(wsi);
 418                 wsi = wsi->same_vh_protocol_next;
 419         }
 420
 421         return 0;
 422 }
 423
 424 /**
 425  * lws_callback_on_writable_all_protocol() - Request a callback for
 426  *                      all connections using the given protocol when it
 427  *                      becomes possible to write to each socket without
 428  *                      blocking in turn.
 429  *
 430  *      This calls back any connection using the same protocol on ANY
 431  *      VHOST.
 432  *
 433  * @context:    lws_context
 434  * @protocol:   Protocol whose connections will get callbacks
 435  */
 436
 437 LWS_VISIBLE int
 438 lws_callback_on_writable_all_protocol(const struct lws_context *context,
 439                                       const struct lws_protocols *protocol)
 440 {
 441         struct lws_vhost *vhost = context->vhost_list;
 442         int n;
 443
 444         while (vhost) {
 445                 for (n = 0; n < vhost->count_protocols; n++)
 446                         if (protocol->callback ==
 447                             vhost->protocols[n].callback &&
 448                             !strcmp(protocol->name, vhost->protocols[n].name))
 449                                 break;
 450                 if (n != vhost->count_protocols)
 451                         lws_callback_on_writable_all_protocol_vhost(
 452                                 vhost, &vhost->protocols[n]);
 453
 454                 vhost = vhost->vhost_next;
 455         }
 456
 457         return 0;
 458 }