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