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 insert_wsi_socket_into_fds(struct lws_context *context, struct lws *wsi)
  26 {
  27         struct lws_pollargs pa = { wsi->sock, LWS_POLLIN, 0 };
  28
  29         if (context->fds_count >= context->max_fds) {
  30                 lwsl_err("Too many fds (%d)\n", context->max_fds);
  31                 return 1;
  32         }
  33
  34 #if !defined(_WIN32) && !defined(MBED_OPERATORS)
  35         if (wsi->sock >= context->max_fds) {
  36                 lwsl_err("Socket fd %d is too high (%d)\n",
  37                          wsi->sock, context->max_fds);
  38                 return 1;
  39         }
  40 #endif
  41
  42         assert(wsi);
  43         assert(lws_socket_is_valid(wsi->sock));
  44
  45         if (context->protocols[0].callback(wsi, LWS_CALLBACK_LOCK_POLL,
  46                                            wsi->user_space, (void *) &pa, 1))
  47                 return -1;
  48
  49         insert_wsi(context, wsi);
  50         wsi->position_in_fds_table = context->fds_count;
  51         context->fds[context->fds_count].fd = wsi->sock;
  52         context->fds[context->fds_count].events = LWS_POLLIN;
  53
  54         lws_plat_insert_socket_into_fds(context, wsi);
  55
  56         /* external POLL support via protocol 0 */
  57         if (context->protocols[0].callback(wsi, LWS_CALLBACK_ADD_POLL_FD,
  58                                            wsi->user_space, (void *) &pa, 0))
  59                 return -1;
  60
  61         if (context->protocols[0].callback(wsi, LWS_CALLBACK_UNLOCK_POLL,
  62                                            wsi->user_space, (void *)&pa, 1))
  63                 return -1;
  64
  65         return 0;
  66 }
  67
  68 int
  69 remove_wsi_socket_from_fds(struct lws *wsi)
  70 {
  71         int m;
  72         struct lws *end_wsi;
  73         struct lws_pollargs pa = { wsi->sock, 0, 0 };
  74         struct lws_context *context = wsi->context;
  75
  76         lws_libev_io(wsi, LWS_EV_STOP | LWS_EV_READ | LWS_EV_WRITE);
  77
  78         --context->fds_count;
  79
  80 #if !defined(_WIN32) && !defined(MBED_OPERATORS)
  81         if (wsi->sock > context->max_fds) {
  82                 lwsl_err("Socket fd %d too high (%d)\n",
  83                          wsi->sock, context->max_fds);
  84                 return 1;
  85         }
  86 #endif
  87
  88         lwsl_info("%s: wsi=%p, sock=%d, fds pos=%d\n", __func__,
  89                   wsi, wsi->sock, wsi->position_in_fds_table);
  90
  91         if (context->protocols[0].callback(wsi, LWS_CALLBACK_LOCK_POLL,
  92                                            wsi->user_space, (void *)&pa, 1))
  93                 return -1;
  94
  95         m = wsi->position_in_fds_table; /* replace the contents for this */
  96
  97         /* have the last guy take up the vacant slot */
  98         context->fds[m] = context->fds[context->fds_count];
  99
 100         lws_plat_delete_socket_from_fds(context, wsi, m);
 101
 102         /*
 103          * end guy's fds_lookup entry remains unchanged
 104          * (still same fd pointing to same wsi)
 105          */
 106         /* end guy's "position in fds table" changed */
 107         end_wsi = wsi_from_fd(context, context->fds[context->fds_count].fd);
 108         end_wsi->position_in_fds_table = m;
 109         /* deletion guy's lws_lookup entry needs nuking */
 110         delete_from_fd(context, wsi->sock);
 111         /* removed wsi has no position any more */
 112         wsi->position_in_fds_table = -1;
 113
 114         /* remove also from external POLL support via protocol 0 */
 115         if (lws_socket_is_valid(wsi->sock)) {
 116                 if (context->protocols[0].callback(wsi, LWS_CALLBACK_DEL_POLL_FD,
 117                     wsi->user_space, (void *) &pa, 0))
 118                         return -1;
 119         }
 120         if (context->protocols[0].callback(wsi, LWS_CALLBACK_UNLOCK_POLL,
 121                                            wsi->user_space, (void *) &pa, 1))
 122                 return -1;
 123
 124         return 0;
 125 }
 126
 127 int
 128 lws_change_pollfd(struct lws *wsi, int _and, int _or)
 129 {
 130         struct lws_context *context;
 131         int tid;
 132         int sampled_tid;
 133         struct lws_pollfd *pfd;
 134         struct lws_pollargs pa;
 135         int pa_events = 1;
 136
 137         if (!wsi || !wsi->protocol || wsi->position_in_fds_table < 0)
 138                 return 1;
 139
 140         context = lws_get_context(wsi);
 141         if (!context)
 142                 return 1;
 143
 144         pfd = &context->fds[wsi->position_in_fds_table];
 145         pa.fd = wsi->sock;
 146
 147         if (context->protocols[0].callback(wsi, LWS_CALLBACK_LOCK_POLL,
 148                                            wsi->user_space,  (void *) &pa, 0))
 149                 return -1;
 150
 151         pa.prev_events = pfd->events;
 152         pa.events = pfd->events = (pfd->events & ~_and) | _or;
 153
 154         if (context->protocols[0].callback(wsi, LWS_CALLBACK_CHANGE_MODE_POLL_FD,
 155                                            wsi->user_space, (void *) &pa, 0))
 156                 return -1;
 157
 158         /*
 159          * if we changed something in this pollfd...
 160          *   ... and we're running in a different thread context
 161          *     than the service thread...
 162          *       ... and the service thread is waiting ...
 163          *         then cancel it to force a restart with our changed events
 164          */
 165 #if LWS_POSIX
 166         pa_events = pa.prev_events != pa.events;
 167 #endif
 168         if (pa_events) {
 169
 170                 if (lws_plat_change_pollfd(context, wsi, pfd)) {
 171                         lwsl_info("%s failed\n", __func__);
 172                         return 1;
 173                 }
 174
 175                 sampled_tid = context->service_tid;
 176                 if (sampled_tid) {
 177                         tid = context->protocols[0].callback(wsi,
 178                                      LWS_CALLBACK_GET_THREAD_ID, NULL, NULL, 0);
 179                         if (tid == -1)
 180                                 return -1;
 181                         if (tid != sampled_tid)
 182                                 lws_cancel_service(context);
 183                 }
 184         }
 185
 186         if (context->protocols[0].callback(wsi, LWS_CALLBACK_UNLOCK_POLL,
 187                                            wsi->user_space, (void *) &pa, 0))
 188                 return -1;
 189
 190         return 0;
 191 }
 192
 193
 194 /**
 195  * lws_callback_on_writable() - Request a callback when this socket
 196  *                                       becomes able to be written to without
 197  *                                       blocking
 198  *
 199  * @wsi:        Websocket connection instance to get callback for
 200  */
 201
 202 LWS_VISIBLE int
 203 lws_callback_on_writable(struct lws *wsi)
 204 {
 205 #ifdef LWS_USE_HTTP2
 206         struct lws *network_wsi, *wsi2;
 207         int already;
 208
 209         lwsl_info("%s: %p\n", __func__, wsi);
 210
 211         if (wsi->mode != LWSCM_HTTP2_SERVING)
 212                 goto network_sock;
 213
 214         if (wsi->u.http2.requested_POLLOUT) {
 215                 lwsl_info("already pending writable\n");
 216                 return 1;
 217         }
 218
 219         if (wsi->u.http2.tx_credit <= 0) {
 220                 /*
 221                  * other side is not able to cope with us sending
 222                  * anything so no matter if we have POLLOUT on our side.
 223                  *
 224                  * Delay waiting for our POLLOUT until peer indicates he has
 225                  * space for more using tx window command in http2 layer
 226                  */
 227                 lwsl_info("%s: %p: waiting_tx_credit (%d)\n", __func__, wsi,
 228                           wsi->u.http2.tx_credit);
 229                 wsi->u.http2.waiting_tx_credit = 1;
 230                 return 0;
 231         }
 232
 233         network_wsi = lws_http2_get_network_wsi(wsi);
 234         already = network_wsi->u.http2.requested_POLLOUT;
 235
 236         /* mark everybody above him as requesting pollout */
 237
 238         wsi2 = wsi;
 239         while (wsi2) {
 240                 wsi2->u.http2.requested_POLLOUT = 1;
 241                 lwsl_info("mark %p pending writable\n", wsi2);
 242                 wsi2 = wsi2->u.http2.parent_wsi;
 243         }
 244
 245         /* for network action, act only on the network wsi */
 246
 247         wsi = network_wsi;
 248         if (already)
 249                 return 1;
 250 network_sock:
 251 #endif
 252
 253         if (lws_ext_cb_active(wsi,
 254                                 LWS_EXT_CB_REQUEST_ON_WRITEABLE, NULL, 0))
 255                 return 1;
 256
 257         if (wsi->position_in_fds_table < 0) {
 258                 lwsl_err("%s: failed to find socket %d\n", __func__, wsi->sock);
 259                 return -1;
 260         }
 261
 262         if (lws_change_pollfd(wsi, 0, LWS_POLLOUT))
 263                 return -1;
 264
 265         lws_libev_io(wsi, LWS_EV_START | LWS_EV_WRITE);
 266
 267         return 1;
 268 }
 269
 270 /**
 271  * lws_callback_on_writable_all_protocol() - Request a callback for
 272  *                      all connections using the given protocol when it
 273  *                      becomes possible to write to each socket without
 274  *                      blocking in turn.
 275  *
 276  * @context:    lws_context
 277  * @protocol:   Protocol whose connections will get callbacks
 278  */
 279
 280 LWS_VISIBLE int
 281 lws_callback_on_writable_all_protocol(const struct lws_context *context,
 282                                       const struct lws_protocols *protocol)
 283 {
 284         struct lws *wsi;
 285         int n;
 286
 287         for (n = 0; n < context->fds_count; n++) {
 288                 wsi = wsi_from_fd(context,context->fds[n].fd);
 289                 if (!wsi)
 290                         continue;
 291                 if (wsi->protocol == protocol)
 292                         lws_callback_on_writable(wsi);
 293         }
 294
 295         return 0;
 296 }