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