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