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         if (wsi->handling_pollout && !_and && _or == LWS_POLLOUT) {
  37                 /*
  38                  * Happening alongside service thread handling POLLOUT.
  39                  * The danger is when he is finished, he will disable POLLOUT,
  40                  * countermanding what we changed here.
  41                  *
  42                  * Instead of changing the fds, inform the service thread
  43                  * what happened, and ask it to leave POLLOUT active on exit
  44                  */
  45                 wsi->leave_pollout_active = 1;
  46                 /*
  47                  * by definition service thread is not in poll wait, so no need
  48                  * to cancel service
  49                  */
  50
  51                 lwsl_debug("%s: using leave_pollout_active\n", __func__);
  52
  53                 return 0;
  54         }
  55
  56         context = wsi->context;
  57         pt = &context->pt[(int)wsi->tsi];
  58         assert(wsi->position_in_fds_table >= 0 &&
  59                wsi->position_in_fds_table < pt->fds_count);
  60
  61         pfd = &pt->fds[wsi->position_in_fds_table];
  62         pa->fd = wsi->desc.sockfd;
  63         pa->prev_events = pfd->events;
  64         pa->events = pfd->events = (pfd->events & ~_and) | _or;
  65
  66         //lwsl_notice("%s: wsi %p, posin %d. from %d -> %d\n", __func__, wsi, wsi->position_in_fds_table, pa->prev_events, pa->events);
  67
  68
  69         if (wsi->http2_substream)
  70                 return 0;
  71
  72         if (wsi->vhost->protocols[0].callback(wsi, LWS_CALLBACK_CHANGE_MODE_POLL_FD,
  73                                            wsi->user_space, (void *)pa, 0)) {
  74                 ret = -1;
  75                 goto bail;
  76         }
  77
  78         if (_and & LWS_POLLIN) {
  79                 lws_libev_io(wsi, LWS_EV_STOP | LWS_EV_READ);
  80                 lws_libuv_io(wsi, LWS_EV_STOP | LWS_EV_READ);
  81                 lws_libevent_io(wsi, LWS_EV_STOP | LWS_EV_READ);
  82         }
  83         if (_or & LWS_POLLIN) {
  84                 lws_libev_io(wsi, LWS_EV_START | LWS_EV_READ);
  85                 lws_libuv_io(wsi, LWS_EV_START | LWS_EV_READ);
  86                 lws_libevent_io(wsi, LWS_EV_START | LWS_EV_READ);
  87         }
  88         if (_and & LWS_POLLOUT) {
  89                 lws_libev_io(wsi, LWS_EV_STOP | LWS_EV_WRITE);
  90                 lws_libuv_io(wsi, LWS_EV_STOP | LWS_EV_WRITE);
  91                 lws_libevent_io(wsi, LWS_EV_STOP | LWS_EV_WRITE);
  92         }
  93         if (_or & LWS_POLLOUT) {
  94                 lws_libev_io(wsi, LWS_EV_START | LWS_EV_WRITE);
  95                 lws_libuv_io(wsi, LWS_EV_START | LWS_EV_WRITE);
  96                 lws_libevent_io(wsi, LWS_EV_START | LWS_EV_WRITE);
  97         }
  98
  99         /*
 100          * if we changed something in this pollfd...
 101          *   ... and we're running in a different thread context
 102          *     than the service thread...
 103          *       ... and the service thread is waiting ...
 104          *         then cancel it to force a restart with our changed events
 105          */
 106 #if LWS_POSIX
 107         pa_events = pa->prev_events != pa->events;
 108 #endif
 109
 110         if (pa_events) {
 111
 112                 if (lws_plat_change_pollfd(context, wsi, pfd)) {
 113                         lwsl_info("%s failed\n", __func__);
 114                         ret = -1;
 115                         goto bail;
 116                 }
 117
 118                 sampled_tid = context->service_tid;
 119                 if (sampled_tid) {
 120                         tid = wsi->vhost->protocols[0].callback(wsi,
 121                                      LWS_CALLBACK_GET_THREAD_ID, NULL, NULL, 0);
 122                         if (tid == -1) {
 123                                 ret = -1;
 124                                 goto bail;
 125                         }
 126                         if (tid != sampled_tid)
 127                                 lws_cancel_service_pt(wsi);
 128                 }
 129         }
 130 bail:
 131         return ret;
 132 }
 133
 134 #ifndef LWS_NO_SERVER
 135 static void
 136 lws_accept_modulation(struct lws_context_per_thread *pt, int allow)
 137 {
 138 // multithread listen seems broken
 139 #if 0
 140         struct lws_vhost *vh = context->vhost_list;
 141         struct lws_pollargs pa1;
 142
 143         while (vh) {
 144                 if (allow)
 145                         _lws_change_pollfd(pt->wsi_listening,
 146                                            0, LWS_POLLIN, &pa1);
 147                 else
 148                         _lws_change_pollfd(pt->wsi_listening,
 149                                            LWS_POLLIN, 0, &pa1);
 150                 vh = vh->vhost_next;
 151         }
 152 #endif
 153 }
 154 #endif
 155
 156 int
 157 insert_wsi_socket_into_fds(struct lws_context *context, struct lws *wsi)
 158 {
 159         struct lws_pollargs pa = { wsi->desc.sockfd, LWS_POLLIN, 0 };
 160         struct lws_context_per_thread *pt = &context->pt[(int)wsi->tsi];
 161         int ret = 0;
 162
 163
 164         lwsl_debug("%s: %p: tsi=%d, sock=%d, pos-in-fds=%d\n",
 165                   __func__, wsi, wsi->tsi, wsi->desc.sockfd, pt->fds_count);
 166
 167         if ((unsigned int)pt->fds_count >= context->fd_limit_per_thread) {
 168                 lwsl_err("Too many fds (%d vs %d)\n", context->max_fds,
 169                                 context->fd_limit_per_thread    );
 170                 return 1;
 171         }
 172
 173 #if !defined(_WIN32) && !defined(LWS_WITH_ESP8266)
 174         if (wsi->desc.sockfd >= context->max_fds) {
 175                 lwsl_err("Socket fd %d is too high (%d)\n",
 176                          wsi->desc.sockfd, context->max_fds);
 177                 return 1;
 178         }
 179 #endif
 180
 181         assert(wsi);
 182         assert(wsi->vhost);
 183         assert(lws_socket_is_valid(wsi->desc.sockfd));
 184
 185         if (wsi->vhost->protocols[0].callback(wsi, LWS_CALLBACK_LOCK_POLL,
 186                                            wsi->user_space, (void *) &pa, 1))
 187                 return -1;
 188
 189         lws_pt_lock(pt);
 190         pt->count_conns++;
 191         insert_wsi(context, wsi);
 192 #if defined(LWS_WITH_ESP8266)
 193         if (wsi->position_in_fds_table == -1)
 194 #endif
 195                 wsi->position_in_fds_table = pt->fds_count;
 196
 197         // lwsl_notice("%s: %p: setting posinfds %d\n", __func__, wsi, wsi->position_in_fds_table);
 198
 199         pt->fds[wsi->position_in_fds_table].fd = wsi->desc.sockfd;
 200 #if LWS_POSIX
 201         pt->fds[wsi->position_in_fds_table].events = LWS_POLLIN;
 202 #else
 203         pt->fds[wsi->position_in_fds_table].events = 0; // LWS_POLLIN;
 204 #endif
 205         pa.events = pt->fds[pt->fds_count].events;
 206
 207         lws_plat_insert_socket_into_fds(context, wsi);
 208
 209         /* external POLL support via protocol 0 */
 210         if (wsi->vhost->protocols[0].callback(wsi, LWS_CALLBACK_ADD_POLL_FD,
 211                                            wsi->user_space, (void *) &pa, 0))
 212                 ret =  -1;
 213 #ifndef LWS_NO_SERVER
 214         /* if no more room, defeat accepts on this thread */
 215         if ((unsigned int)pt->fds_count == context->fd_limit_per_thread - 1)
 216                 lws_accept_modulation(pt, 0);
 217 #endif
 218         lws_pt_unlock(pt);
 219
 220         if (wsi->vhost->protocols[0].callback(wsi, LWS_CALLBACK_UNLOCK_POLL,
 221                                            wsi->user_space, (void *)&pa, 1))
 222                 ret = -1;
 223
 224         return ret;
 225 }
 226
 227 int
 228 remove_wsi_socket_from_fds(struct lws *wsi)
 229 {
 230         struct lws_context *context = wsi->context;
 231         struct lws_pollargs pa = { wsi->desc.sockfd, 0, 0 };
 232 #if !defined(LWS_WITH_ESP8266)
 233         struct lws_context_per_thread *pt = &context->pt[(int)wsi->tsi];
 234         struct lws *end_wsi;
 235         int v;
 236 #endif
 237         int m, ret = 0;
 238
 239         if (wsi->parent_carries_io) {
 240                 lws_same_vh_protocol_remove(wsi);
 241                 return 0;
 242         }
 243
 244 #if !defined(_WIN32) && !defined(LWS_WITH_ESP8266)
 245         if (wsi->desc.sockfd > context->max_fds) {
 246                 lwsl_err("fd %d too high (%d)\n", wsi->desc.sockfd, context->max_fds);
 247                 return 1;
 248         }
 249 #endif
 250
 251         if (wsi->vhost->protocols[0].callback(wsi, LWS_CALLBACK_LOCK_POLL,
 252                                            wsi->user_space, (void *)&pa, 1))
 253                 return -1;
 254
 255         lws_same_vh_protocol_remove(wsi);
 256
 257         /* the guy who is to be deleted's slot index in pt->fds */
 258         m = wsi->position_in_fds_table;
 259
 260 #if !defined(LWS_WITH_ESP8266)
 261         lws_libev_io(wsi, LWS_EV_STOP | LWS_EV_READ | LWS_EV_WRITE | LWS_EV_PREPARE_DELETION);
 262         lws_libuv_io(wsi, LWS_EV_STOP | LWS_EV_READ | LWS_EV_WRITE | LWS_EV_PREPARE_DELETION);
 263
 264         lws_pt_lock(pt);
 265
 266         lwsl_debug("%s: wsi=%p, sock=%d, fds pos=%d, end guy pos=%d, endfd=%d\n",
 267                   __func__, wsi, wsi->desc.sockfd, wsi->position_in_fds_table,
 268                   pt->fds_count, pt->fds[pt->fds_count].fd);
 269
 270         /* have the last guy take up the now vacant slot */
 271         pt->fds[m] = pt->fds[pt->fds_count - 1];
 272 #endif
 273         /* this decrements pt->fds_count */
 274         lws_plat_delete_socket_from_fds(context, wsi, m);
 275 #if !defined(LWS_WITH_ESP8266)
 276         v = (int) pt->fds[m].fd;
 277         /* end guy's "position in fds table" is now the deletion guy's old one */
 278         end_wsi = wsi_from_fd(context, v);
 279         if (!end_wsi) {
 280                 lwsl_err("no wsi found for sock fd %d at pos %d, pt->fds_count=%d\n", (int)pt->fds[m].fd, m, pt->fds_count);
 281                 assert(0);
 282         } else
 283                 end_wsi->position_in_fds_table = m;
 284
 285         /* deletion guy's lws_lookup entry needs nuking */
 286         delete_from_fd(context, wsi->desc.sockfd);
 287         /* removed wsi has no position any more */
 288         wsi->position_in_fds_table = -1;
 289
 290         /* remove also from external POLL support via protocol 0 */
 291         if (lws_socket_is_valid(wsi->desc.sockfd))
 292                 if (wsi->vhost->protocols[0].callback(wsi, LWS_CALLBACK_DEL_POLL_FD,
 293                                                    wsi->user_space, (void *) &pa, 0))
 294                         ret = -1;
 295 #ifndef LWS_NO_SERVER
 296         if (!context->being_destroyed)
 297                 /* if this made some room, accept connects on this thread */
 298                 if ((unsigned int)pt->fds_count < context->fd_limit_per_thread - 1)
 299                         lws_accept_modulation(pt, 1);
 300 #endif
 301         lws_pt_unlock(pt);
 302
 303         if (wsi->vhost->protocols[0].callback(wsi, LWS_CALLBACK_UNLOCK_POLL,
 304                                            wsi->user_space, (void *) &pa, 1))
 305                 ret = -1;
 306 #endif
 307         return ret;
 308 }
 309
 310 int
 311 lws_change_pollfd(struct lws *wsi, int _and, int _or)
 312 {
 313         struct lws_context_per_thread *pt;
 314         struct lws_context *context;
 315         struct lws_pollargs pa;
 316         int ret = 0;
 317
 318         if (!wsi || !wsi->protocol || wsi->position_in_fds_table < 0)
 319                 return 1;
 320
 321         context = lws_get_context(wsi);
 322         if (!context)
 323                 return 1;
 324
 325         if (wsi->vhost->protocols[0].callback(wsi, LWS_CALLBACK_LOCK_POLL,
 326                                            wsi->user_space,  (void *) &pa, 0))
 327                 return -1;
 328
 329         pt = &context->pt[(int)wsi->tsi];
 330
 331         lws_pt_lock(pt);
 332         ret = _lws_change_pollfd(wsi, _and, _or, &pa);
 333         lws_pt_unlock(pt);
 334         if (wsi->vhost->protocols[0].callback(wsi, LWS_CALLBACK_UNLOCK_POLL,
 335                                            wsi->user_space, (void *) &pa, 0))
 336                 ret = -1;
 337
 338         return ret;
 339 }
 340
 341 LWS_VISIBLE int
 342 lws_callback_on_writable(struct lws *wsi)
 343 {
 344         struct lws_context_per_thread *pt;
 345 #ifdef LWS_USE_HTTP2
 346         struct lws *network_wsi, *wsi2;
 347         int already;
 348 #endif
 349
 350         if (wsi->state == LWSS_SHUTDOWN)
 351                 return 0;
 352
 353         if (wsi->socket_is_permanently_unusable)
 354                 return 0;
 355
 356         if (wsi->parent_carries_io) {
 357                 int n = lws_callback_on_writable(wsi->parent);
 358
 359                 if (n < 0)
 360                         return n;
 361
 362                 wsi->parent_pending_cb_on_writable = 1;
 363                 return 1;
 364         }
 365
 366         pt = &wsi->context->pt[(int)wsi->tsi];
 367         lws_stats_atomic_bump(wsi->context, pt, LWSSTATS_C_WRITEABLE_CB_REQ, 1);
 368 #if defined(LWS_WITH_STATS)
 369         if (!wsi->active_writable_req_us) {
 370                 wsi->active_writable_req_us = time_in_microseconds();
 371                 lws_stats_atomic_bump(wsi->context, pt, LWSSTATS_C_WRITEABLE_CB_EFF_REQ, 1);
 372         }
 373 #endif
 374
 375 #ifdef LWS_USE_HTTP2
 376         lwsl_info("%s: %p\n", __func__, wsi);
 377
 378         if (wsi->mode != LWSCM_HTTP2_SERVING)
 379                 goto network_sock;
 380
 381         if (wsi->u.http2.requested_POLLOUT) {
 382                 lwsl_info("already pending writable\n");
 383                 return 1;
 384         }
 385
 386         if (wsi->u.http2.tx_credit <= 0) {
 387                 /*
 388                  * other side is not able to cope with us sending
 389                  * anything so no matter if we have POLLOUT on our side.
 390                  *
 391                  * Delay waiting for our POLLOUT until peer indicates he has
 392                  * space for more using tx window command in http2 layer
 393                  */
 394                 lwsl_info("%s: %p: waiting_tx_credit (%d)\n", __func__, wsi,
 395                           wsi->u.http2.tx_credit);
 396                 wsi->u.http2.waiting_tx_credit = 1;
 397                 return 0;
 398         }
 399
 400         network_wsi = lws_http2_get_network_wsi(wsi);
 401         already = network_wsi->u.http2.requested_POLLOUT;
 402
 403         /* mark everybody above him as requesting pollout */
 404
 405         wsi2 = wsi;
 406         while (wsi2) {
 407                 wsi2->u.http2.requested_POLLOUT = 1;
 408                 lwsl_info("mark %p pending writable\n", wsi2);
 409                 wsi2 = wsi2->u.http2.parent_wsi;
 410         }
 411
 412         /* for network action, act only on the network wsi */
 413
 414         wsi = network_wsi;
 415         if (already)
 416                 return 1;
 417 network_sock:
 418 #endif
 419
 420         if (lws_ext_cb_active(wsi, LWS_EXT_CB_REQUEST_ON_WRITEABLE, NULL, 0))
 421                 return 1;
 422
 423         if (wsi->position_in_fds_table < 0) {
 424                 lwsl_err("%s: failed to find socket %d\n", __func__, wsi->desc.sockfd);
 425                 return -1;
 426         }
 427
 428         if (lws_change_pollfd(wsi, 0, LWS_POLLOUT))
 429                 return -1;
 430
 431         return 1;
 432 }
 433
 434 /*
 435  * stitch protocol choice into the vh protocol linked list
 436  * We always insert ourselves at the start of the list
 437  *
 438  * X <-> B
 439  * X <-> pAn <-> pB
 440  *
 441  * Illegal to attach more than once without detach inbetween
 442  */
 443 void
 444 lws_same_vh_protocol_insert(struct lws *wsi, int n)
 445 {
 446         //lwsl_err("%s: pre insert vhost start wsi %p, that wsi prev == %p\n",
 447         //              __func__,
 448         //              wsi->vhost->same_vh_protocol_list[n],
 449         //              wsi->same_vh_protocol_prev);
 450
 451         if (wsi->same_vh_protocol_prev || wsi->same_vh_protocol_next) {
 452                 lwsl_err("Attempted to attach wsi twice to same vh prot\n");
 453                 assert(0);
 454         }
 455
 456         wsi->same_vh_protocol_prev = /* guy who points to us */
 457                 &wsi->vhost->same_vh_protocol_list[n];
 458         wsi->same_vh_protocol_next = /* old first guy is our next */
 459                         wsi->vhost->same_vh_protocol_list[n];
 460         /* we become the new first guy */
 461         wsi->vhost->same_vh_protocol_list[n] = wsi;
 462
 463         if (wsi->same_vh_protocol_next)
 464                 /* old first guy points back to us now */
 465                 wsi->same_vh_protocol_next->same_vh_protocol_prev =
 466                                 &wsi->same_vh_protocol_next;
 467 }
 468
 469 void
 470 lws_same_vh_protocol_remove(struct lws *wsi)
 471 {
 472         /*
 473          * detach ourselves from vh protocol list if we're on one
 474          * A -> B -> C
 475          * A -> C , or, B -> C, or A -> B
 476          *
 477          * OK to call on already-detached wsi
 478          */
 479         lwsl_info("%s: removing same prot wsi %p\n", __func__, wsi);
 480
 481         if (wsi->same_vh_protocol_prev) {
 482                 assert (*(wsi->same_vh_protocol_prev) == wsi);
 483                 lwsl_info("have prev %p, setting him to our next %p\n",
 484                          wsi->same_vh_protocol_prev,
 485                          wsi->same_vh_protocol_next);
 486
 487                 /* guy who pointed to us should point to our next */
 488                 *(wsi->same_vh_protocol_prev) = wsi->same_vh_protocol_next;
 489         }
 490
 491         /* our next should point back to our prev */
 492         if (wsi->same_vh_protocol_next) {
 493                 wsi->same_vh_protocol_next->same_vh_protocol_prev =
 494                                 wsi->same_vh_protocol_prev;
 495         }
 496
 497         wsi->same_vh_protocol_prev = NULL;
 498         wsi->same_vh_protocol_next = NULL;
 499 }
 500
 501
 502 LWS_VISIBLE int
 503 lws_callback_on_writable_all_protocol_vhost(const struct lws_vhost *vhost,
 504                                       const struct lws_protocols *protocol)
 505 {
 506         struct lws *wsi;
 507
 508         if (protocol < vhost->protocols ||
 509             protocol >= (vhost->protocols + vhost->count_protocols)) {
 510
 511                 lwsl_err("%s: protocol %p is not from vhost %p (%p - %p)\n",
 512                         __func__, protocol, vhost->protocols, vhost,
 513                         (vhost->protocols + vhost->count_protocols));
 514
 515                 return -1;
 516         }
 517
 518         wsi = vhost->same_vh_protocol_list[protocol - vhost->protocols];
 519         //lwsl_notice("%s: protocol %p, start wsi %p\n", __func__, protocol, wsi);
 520         while (wsi) {
 521                 //lwsl_notice("%s: protocol %p, this wsi %p (wsi->protocol=%p)\n",
 522                 //              __func__, protocol, wsi, wsi->protocol);
 523                 assert(wsi->protocol == protocol);
 524                 assert(*wsi->same_vh_protocol_prev == wsi);
 525                 if (wsi->same_vh_protocol_next) {
 526                 //      lwsl_err("my next says %p\n", wsi->same_vh_protocol_next);
 527                 //      lwsl_err("my next's prev says %p\n",
 528                 //              wsi->same_vh_protocol_next->same_vh_protocol_prev);
 529                         assert(wsi->same_vh_protocol_next->same_vh_protocol_prev == &wsi->same_vh_protocol_next);
 530                 }
 531                 //lwsl_notice("  apv: %p\n", wsi);
 532                 lws_callback_on_writable(wsi);
 533                 wsi = wsi->same_vh_protocol_next;
 534         }
 535
 536         return 0;
 537 }
 538
 539 LWS_VISIBLE int
 540 lws_callback_on_writable_all_protocol(const struct lws_context *context,
 541                                       const struct lws_protocols *protocol)
 542 {
 543         struct lws_vhost *vhost = context->vhost_list;
 544         int n;
 545
 546         while (vhost) {
 547                 for (n = 0; n < vhost->count_protocols; n++)
 548                         if (protocol->callback ==
 549                             vhost->protocols[n].callback &&
 550                             !strcmp(protocol->name, vhost->protocols[n].name))
 551                                 break;
 552                 if (n != vhost->count_protocols)
 553                         lws_callback_on_writable_all_protocol_vhost(
 554                                 vhost, &vhost->protocols[n]);
 555
 556                 vhost = vhost->vhost_next;
 557         }
 558
 559         return 0;
 560 }