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         if (wsi->vhost->protocols[0].callback(wsi, LWS_CALLBACK_CHANGE_MODE_POLL_FD,
  47                                            wsi->user_space, (void *)pa, 0)) {
  48                 ret = -1;
  49                 goto bail;
  50         }
  51
  52         if (_and & LWS_POLLIN) {
  53                 lws_libev_io(wsi, LWS_EV_STOP | LWS_EV_READ);
  54                 lws_libuv_io(wsi, LWS_EV_STOP | LWS_EV_READ);
  55         }
  56         if (_or & LWS_POLLIN) {
  57                 lws_libev_io(wsi, LWS_EV_START | LWS_EV_READ);
  58                 lws_libuv_io(wsi, LWS_EV_START | LWS_EV_READ);
  59         }
  60         if (_and & LWS_POLLOUT) {
  61                 lws_libev_io(wsi, LWS_EV_STOP | LWS_EV_WRITE);
  62                 lws_libuv_io(wsi, LWS_EV_STOP | LWS_EV_WRITE);
  63         }
  64         if (_or & LWS_POLLOUT) {
  65                 lws_libev_io(wsi, LWS_EV_START | LWS_EV_WRITE);
  66                 lws_libuv_io(wsi, LWS_EV_START | LWS_EV_WRITE);
  67         }
  68
  69         /*
  70          * if we changed something in this pollfd...
  71          *   ... and we're running in a different thread context
  72          *     than the service thread...
  73          *       ... and the service thread is waiting ...
  74          *         then cancel it to force a restart with our changed events
  75          */
  76 #if LWS_POSIX
  77         pa_events = pa->prev_events != pa->events;
  78 #endif
  79         if (pa_events) {
  80
  81                 if (lws_plat_change_pollfd(context, wsi, pfd)) {
  82                         lwsl_info("%s failed\n", __func__);
  83                         ret = -1;
  84                         goto bail;
  85                 }
  86
  87                 sampled_tid = context->service_tid;
  88                 if (sampled_tid) {
  89                         tid = wsi->vhost->protocols[0].callback(wsi,
  90                                      LWS_CALLBACK_GET_THREAD_ID, NULL, NULL, 0);
  91                         if (tid == -1) {
  92                                 ret = -1;
  93                                 goto bail;
  94                         }
  95                         if (tid != sampled_tid)
  96                                 lws_cancel_service_pt(wsi);
  97                 }
  98         }
  99 bail:
 100         return ret;
 101 }
 102
 103 int
 104 insert_wsi_socket_into_fds(struct lws_context *context, struct lws *wsi)
 105 {
 106         struct lws_pollargs pa = { wsi->sock, LWS_POLLIN, 0 };
 107         struct lws_context_per_thread *pt = &context->pt[(int)wsi->tsi];
 108         int ret = 0;
 109 #ifndef LWS_NO_SERVER
 110         struct lws_pollargs pa1;
 111 #endif
 112
 113         lwsl_debug("%s: %p: tsi=%d, sock=%d, pos-in-fds=%d\n",
 114                   __func__, wsi, wsi->tsi, wsi->sock, pt->fds_count);
 115
 116         if ((unsigned int)pt->fds_count >= context->fd_limit_per_thread) {
 117                 lwsl_err("Too many fds (%d vs %d)\n", context->max_fds,
 118                                 context->fd_limit_per_thread    );
 119                 return 1;
 120         }
 121
 122 #if !defined(_WIN32) && !defined(MBED_OPERATORS)
 123         if (wsi->sock >= context->max_fds) {
 124                 lwsl_err("Socket fd %d is too high (%d)\n",
 125                          wsi->sock, context->max_fds);
 126                 return 1;
 127         }
 128 #endif
 129
 130         assert(wsi);
 131         assert(wsi->vhost);
 132         assert(lws_socket_is_valid(wsi->sock));
 133
 134         if (wsi->vhost->protocols[0].callback(wsi, LWS_CALLBACK_LOCK_POLL,
 135                                            wsi->user_space, (void *) &pa, 1))
 136                 return -1;
 137
 138         lws_pt_lock(pt);
 139         pt->count_conns++;
 140         insert_wsi(context, wsi);
 141         wsi->position_in_fds_table = pt->fds_count;
 142         pt->fds[pt->fds_count].fd = wsi->sock;
 143         pt->fds[pt->fds_count].events = LWS_POLLIN;
 144         pa.events = pt->fds[pt->fds_count].events;
 145
 146         lws_plat_insert_socket_into_fds(context, wsi);
 147
 148         /* external POLL support via protocol 0 */
 149         if (wsi->vhost->protocols[0].callback(wsi, LWS_CALLBACK_ADD_POLL_FD,
 150                                            wsi->user_space, (void *) &pa, 0))
 151                 ret =  -1;
 152 #ifndef LWS_NO_SERVER
 153         /* if no more room, defeat accepts on this thread */
 154         if ((unsigned int)pt->fds_count == context->fd_limit_per_thread - 1)
 155                 _lws_change_pollfd(pt->wsi_listening, LWS_POLLIN, 0, &pa1);
 156 #endif
 157         lws_pt_unlock(pt);
 158
 159         if (wsi->vhost->protocols[0].callback(wsi, LWS_CALLBACK_UNLOCK_POLL,
 160                                            wsi->user_space, (void *)&pa, 1))
 161                 ret = -1;
 162
 163         return ret;
 164 }
 165
 166 int
 167 remove_wsi_socket_from_fds(struct lws *wsi)
 168 {
 169         struct lws_pollargs pa = { wsi->sock, 0, 0 };
 170 #ifndef LWS_NO_SERVER
 171         struct lws_pollargs pa1;
 172 #endif
 173         struct lws_context *context = wsi->context;
 174         struct lws_context_per_thread *pt = &context->pt[(int)wsi->tsi];
 175         struct lws *end_wsi;
 176         int m, ret = 0;
 177
 178 #if !defined(_WIN32) && !defined(MBED_OPERATORS)
 179         if (wsi->sock > context->max_fds) {
 180                 lwsl_err("fd %d too high (%d)\n", wsi->sock, context->max_fds);
 181                 return 1;
 182         }
 183 #endif
 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_libev_io(wsi, LWS_EV_STOP | LWS_EV_READ | LWS_EV_WRITE | LWS_EV_PREPARE_DELETION);
 190         lws_libuv_io(wsi, LWS_EV_STOP | LWS_EV_READ | LWS_EV_WRITE | LWS_EV_PREPARE_DELETION);
 191
 192         lws_pt_lock(pt);
 193
 194         lwsl_info("%s: wsi=%p, sock=%d, fds pos=%d, end guy pos=%d, endfd=%d\n",
 195                   __func__, wsi, wsi->sock, wsi->position_in_fds_table,
 196                   pt->fds_count, pt->fds[pt->fds_count].fd);
 197
 198         /* the guy who is to be deleted's slot index in pt->fds */
 199         m = wsi->position_in_fds_table;
 200
 201         /* have the last guy take up the now vacant slot */
 202         pt->fds[m] = pt->fds[pt->fds_count - 1];
 203
 204         lws_plat_delete_socket_from_fds(context, wsi, m);
 205
 206         /* end guy's "position in fds table" is now the deletion guy's old one */
 207         end_wsi = wsi_from_fd(context, pt->fds[pt->fds_count].fd);
 208         assert(end_wsi);
 209         end_wsi->position_in_fds_table = m;
 210
 211         /* deletion guy's lws_lookup entry needs nuking */
 212         delete_from_fd(context, wsi->sock);
 213         /* removed wsi has no position any more */
 214         wsi->position_in_fds_table = -1;
 215
 216         /* remove also from external POLL support via protocol 0 */
 217         if (lws_socket_is_valid(wsi->sock))
 218                 if (wsi->vhost->protocols[0].callback(wsi, LWS_CALLBACK_DEL_POLL_FD,
 219                                                    wsi->user_space, (void *) &pa, 0))
 220                         ret = -1;
 221 #ifndef LWS_NO_SERVER
 222         if (!context->being_destroyed)
 223                 /* if this made some room, accept connects on this thread */
 224                 if ((unsigned int)pt->fds_count < context->fd_limit_per_thread - 1)
 225                         _lws_change_pollfd(pt->wsi_listening, 0, LWS_POLLIN, &pa1);
 226 #endif
 227         lws_pt_unlock(pt);
 228
 229         if (wsi->vhost->protocols[0].callback(wsi, LWS_CALLBACK_UNLOCK_POLL,
 230                                            wsi->user_space, (void *) &pa, 1))
 231                 ret = -1;
 232
 233         return ret;
 234 }
 235
 236 int
 237 lws_change_pollfd(struct lws *wsi, int _and, int _or)
 238 {
 239         struct lws_context_per_thread *pt;
 240         struct lws_context *context;
 241         struct lws_pollargs pa;
 242         int ret = 0;
 243
 244         if (!wsi || !wsi->protocol || wsi->position_in_fds_table < 0)
 245                 return 1;
 246
 247         context = lws_get_context(wsi);
 248         if (!context)
 249                 return 1;
 250
 251         if (wsi->vhost->protocols[0].callback(wsi, LWS_CALLBACK_LOCK_POLL,
 252                                            wsi->user_space,  (void *) &pa, 0))
 253                 return -1;
 254
 255         pt = &context->pt[(int)wsi->tsi];
 256
 257         lws_pt_lock(pt);
 258         ret = _lws_change_pollfd(wsi, _and, _or, &pa);
 259         lws_pt_unlock(pt);
 260         if (wsi->vhost->protocols[0].callback(wsi, LWS_CALLBACK_UNLOCK_POLL,
 261                                            wsi->user_space, (void *) &pa, 0))
 262                 ret = -1;
 263
 264         return ret;
 265 }
 266
 267
 268 /**
 269  * lws_callback_on_writable() - Request a callback when this socket
 270  *                                       becomes able to be written to without
 271  *                                       blocking
 272  *
 273  * @wsi:        Websocket connection instance to get callback for
 274  */
 275
 276 LWS_VISIBLE int
 277 lws_callback_on_writable(struct lws *wsi)
 278 {
 279 #ifdef LWS_USE_HTTP2
 280         struct lws *network_wsi, *wsi2;
 281         int already;
 282 #endif
 283
 284         if (wsi->state == LWSS_SHUTDOWN)
 285                 return 0;
 286
 287 #ifdef LWS_USE_HTTP2
 288         lwsl_info("%s: %p\n", __func__, wsi);
 289
 290         if (wsi->mode != LWSCM_HTTP2_SERVING)
 291                 goto network_sock;
 292
 293         if (wsi->u.http2.requested_POLLOUT) {
 294                 lwsl_info("already pending writable\n");
 295                 return 1;
 296         }
 297
 298         if (wsi->u.http2.tx_credit <= 0) {
 299                 /*
 300                  * other side is not able to cope with us sending
 301                  * anything so no matter if we have POLLOUT on our side.
 302                  *
 303                  * Delay waiting for our POLLOUT until peer indicates he has
 304                  * space for more using tx window command in http2 layer
 305                  */
 306                 lwsl_info("%s: %p: waiting_tx_credit (%d)\n", __func__, wsi,
 307                           wsi->u.http2.tx_credit);
 308                 wsi->u.http2.waiting_tx_credit = 1;
 309                 return 0;
 310         }
 311
 312         network_wsi = lws_http2_get_network_wsi(wsi);
 313         already = network_wsi->u.http2.requested_POLLOUT;
 314
 315         /* mark everybody above him as requesting pollout */
 316
 317         wsi2 = wsi;
 318         while (wsi2) {
 319                 wsi2->u.http2.requested_POLLOUT = 1;
 320                 lwsl_info("mark %p pending writable\n", wsi2);
 321                 wsi2 = wsi2->u.http2.parent_wsi;
 322         }
 323
 324         /* for network action, act only on the network wsi */
 325
 326         wsi = network_wsi;
 327         if (already)
 328                 return 1;
 329 network_sock:
 330 #endif
 331
 332         if (lws_ext_cb_active(wsi, LWS_EXT_CB_REQUEST_ON_WRITEABLE, NULL, 0))
 333                 return 1;
 334
 335         if (wsi->position_in_fds_table < 0) {
 336                 lwsl_err("%s: failed to find socket %d\n", __func__, wsi->sock);
 337                 return -1;
 338         }
 339
 340         if (lws_change_pollfd(wsi, 0, LWS_POLLOUT))
 341                 return -1;
 342
 343         return 1;
 344 }
 345
 346 /**
 347  * lws_callback_on_writable_all_protocol() - Request a callback for
 348  *                      all connections using the given protocol when it
 349  *                      becomes possible to write to each socket without
 350  *                      blocking in turn.
 351  *
 352  * @context:    lws_context
 353  * @protocol:   Protocol whose connections will get callbacks
 354  */
 355
 356 LWS_VISIBLE int
 357 lws_callback_on_writable_all_protocol(const struct lws_context *context,
 358                                       const struct lws_protocols *protocol)
 359 {
 360         const struct lws_context_per_thread *pt = &context->pt[0];
 361         unsigned int n, m = context->count_threads;
 362         struct lws *wsi;
 363
 364         while (m--) {
 365                 for (n = 0; n < pt->fds_count; n++) {
 366                         wsi = wsi_from_fd(context, pt->fds[n].fd);
 367                         if (!wsi)
 368                                 continue;
 369                         if (wsi->protocol == protocol)
 370                                 lws_callback_on_writable(wsi);
 371                 }
 372                 pt++;
 373         }
 374
 375         return 0;
 376 }