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