lib/pollfd.c

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