if (!wsi || wsi->position_in_fds_table < 0)
return 0;
+ if (wsi->handling_pollout && !_and && _or == LWS_POLLOUT) {
+ /*
+ * Happening alongside service thread handling POLLOUT.
+ * The danger is when he is finished, he will disable POLLOUT,
+ * countermanding what we changed here.
+ *
+ * Instead of changing the fds, inform the service thread
+ * what happened, and ask it to leave POLLOUT active on exit
+ */
+ wsi->leave_pollout_active = 1;
+ /*
+ * by definition service thread is not in poll wait, so no need
+ * to cancel service
+ */
+
+ lwsl_debug("%s: using leave_pollout_active\n", __func__);
+
+ return 0;
+ }
+
context = wsi->context;
pt = &context->pt[(int)wsi->tsi];
assert(wsi->position_in_fds_table >= 0 &&
wsi->position_in_fds_table < pt->fds_count);
pfd = &pt->fds[wsi->position_in_fds_table];
- pa->fd = wsi->sock;
+ pa->fd = wsi->desc.sockfd;
pa->prev_events = pfd->events;
pa->events = pfd->events = (pfd->events & ~_and) | _or;
- if (context->protocols[0].callback(wsi, LWS_CALLBACK_CHANGE_MODE_POLL_FD,
+ //lwsl_notice("%s: wsi %p, posin %d. from %d -> %d\n", __func__, wsi, wsi->position_in_fds_table, pa->prev_events, pa->events);
+
+
+ if (wsi->http2_substream)
+ return 0;
+
+ if (wsi->vhost->protocols[0].callback(wsi, LWS_CALLBACK_CHANGE_MODE_POLL_FD,
wsi->user_space, (void *)pa, 0)) {
ret = -1;
goto bail;
if (_and & LWS_POLLIN) {
lws_libev_io(wsi, LWS_EV_STOP | LWS_EV_READ);
lws_libuv_io(wsi, LWS_EV_STOP | LWS_EV_READ);
+ lws_libevent_io(wsi, LWS_EV_STOP | LWS_EV_READ);
}
if (_or & LWS_POLLIN) {
lws_libev_io(wsi, LWS_EV_START | LWS_EV_READ);
lws_libuv_io(wsi, LWS_EV_START | LWS_EV_READ);
+ lws_libevent_io(wsi, LWS_EV_START | LWS_EV_READ);
}
if (_and & LWS_POLLOUT) {
lws_libev_io(wsi, LWS_EV_STOP | LWS_EV_WRITE);
lws_libuv_io(wsi, LWS_EV_STOP | LWS_EV_WRITE);
+ lws_libevent_io(wsi, LWS_EV_STOP | LWS_EV_WRITE);
}
if (_or & LWS_POLLOUT) {
lws_libev_io(wsi, LWS_EV_START | LWS_EV_WRITE);
lws_libuv_io(wsi, LWS_EV_START | LWS_EV_WRITE);
+ lws_libevent_io(wsi, LWS_EV_START | LWS_EV_WRITE);
}
/*
#if LWS_POSIX
pa_events = pa->prev_events != pa->events;
#endif
+
if (pa_events) {
if (lws_plat_change_pollfd(context, wsi, pfd)) {
sampled_tid = context->service_tid;
if (sampled_tid) {
- tid = context->protocols[0].callback(wsi,
+ tid = wsi->vhost->protocols[0].callback(wsi,
LWS_CALLBACK_GET_THREAD_ID, NULL, NULL, 0);
if (tid == -1) {
ret = -1;
return ret;
}
+#ifndef LWS_NO_SERVER
+static void
+lws_accept_modulation(struct lws_context_per_thread *pt, int allow)
+{
+// multithread listen seems broken
+#if 0
+ struct lws_vhost *vh = context->vhost_list;
+ struct lws_pollargs pa1;
+
+ while (vh) {
+ if (allow)
+ _lws_change_pollfd(pt->wsi_listening,
+ 0, LWS_POLLIN, &pa1);
+ else
+ _lws_change_pollfd(pt->wsi_listening,
+ LWS_POLLIN, 0, &pa1);
+ vh = vh->vhost_next;
+ }
+#endif
+}
+#endif
+
int
insert_wsi_socket_into_fds(struct lws_context *context, struct lws *wsi)
{
- struct lws_pollargs pa = { wsi->sock, LWS_POLLIN, 0 };
+ struct lws_pollargs pa = { wsi->desc.sockfd, LWS_POLLIN, 0 };
struct lws_context_per_thread *pt = &context->pt[(int)wsi->tsi];
int ret = 0;
-#ifndef LWS_NO_SERVER
- struct lws_pollargs pa1;
-#endif
+
lwsl_debug("%s: %p: tsi=%d, sock=%d, pos-in-fds=%d\n",
- __func__, wsi, wsi->tsi, wsi->sock, pt->fds_count);
+ __func__, wsi, wsi->tsi, wsi->desc.sockfd, pt->fds_count);
if ((unsigned int)pt->fds_count >= context->fd_limit_per_thread) {
- lwsl_err("Too many fds (%d)\n", context->max_fds);
+ lwsl_err("Too many fds (%d vs %d)\n", context->max_fds,
+ context->fd_limit_per_thread );
return 1;
}
-#if !defined(_WIN32) && !defined(MBED_OPERATORS)
- if (wsi->sock >= context->max_fds) {
+#if !defined(_WIN32) && !defined(LWS_WITH_ESP8266)
+ if (wsi->desc.sockfd >= context->max_fds) {
lwsl_err("Socket fd %d is too high (%d)\n",
- wsi->sock, context->max_fds);
+ wsi->desc.sockfd, context->max_fds);
return 1;
}
#endif
assert(wsi);
- assert(lws_socket_is_valid(wsi->sock));
+ assert(wsi->vhost);
+ assert(lws_socket_is_valid(wsi->desc.sockfd));
- if (context->protocols[0].callback(wsi, LWS_CALLBACK_LOCK_POLL,
+ if (wsi->vhost->protocols[0].callback(wsi, LWS_CALLBACK_LOCK_POLL,
wsi->user_space, (void *) &pa, 1))
return -1;
lws_pt_lock(pt);
pt->count_conns++;
insert_wsi(context, wsi);
- wsi->position_in_fds_table = pt->fds_count;
- pt->fds[pt->fds_count].fd = wsi->sock;
- pt->fds[pt->fds_count].events = LWS_POLLIN;
+#if defined(LWS_WITH_ESP8266)
+ if (wsi->position_in_fds_table == -1)
+#endif
+ wsi->position_in_fds_table = pt->fds_count;
+
+ // lwsl_notice("%s: %p: setting posinfds %d\n", __func__, wsi, wsi->position_in_fds_table);
+
+ pt->fds[wsi->position_in_fds_table].fd = wsi->desc.sockfd;
+#if LWS_POSIX
+ pt->fds[wsi->position_in_fds_table].events = LWS_POLLIN;
+#else
+ pt->fds[wsi->position_in_fds_table].events = 0; // LWS_POLLIN;
+#endif
pa.events = pt->fds[pt->fds_count].events;
lws_plat_insert_socket_into_fds(context, wsi);
/* external POLL support via protocol 0 */
- if (context->protocols[0].callback(wsi, LWS_CALLBACK_ADD_POLL_FD,
+ if (wsi->vhost->protocols[0].callback(wsi, LWS_CALLBACK_ADD_POLL_FD,
wsi->user_space, (void *) &pa, 0))
ret = -1;
#ifndef LWS_NO_SERVER
/* if no more room, defeat accepts on this thread */
if ((unsigned int)pt->fds_count == context->fd_limit_per_thread - 1)
- _lws_change_pollfd(pt->wsi_listening, LWS_POLLIN, 0, &pa1);
+ lws_accept_modulation(pt, 0);
#endif
lws_pt_unlock(pt);
- if (context->protocols[0].callback(wsi, LWS_CALLBACK_UNLOCK_POLL,
+ if (wsi->vhost->protocols[0].callback(wsi, LWS_CALLBACK_UNLOCK_POLL,
wsi->user_space, (void *)&pa, 1))
ret = -1;
int
remove_wsi_socket_from_fds(struct lws *wsi)
{
- struct lws_pollargs pa = { wsi->sock, 0, 0 };
-#ifndef LWS_NO_SERVER
- struct lws_pollargs pa1;
-#endif
struct lws_context *context = wsi->context;
+ struct lws_pollargs pa = { wsi->desc.sockfd, 0, 0 };
+#if !defined(LWS_WITH_ESP8266)
struct lws_context_per_thread *pt = &context->pt[(int)wsi->tsi];
struct lws *end_wsi;
+ int v;
+#endif
int m, ret = 0;
-#if !defined(_WIN32) && !defined(MBED_OPERATORS)
- if (wsi->sock > context->max_fds) {
- lwsl_err("fd %d too high (%d)\n", wsi->sock, context->max_fds);
+ if (wsi->parent_carries_io) {
+ lws_same_vh_protocol_remove(wsi);
+ return 0;
+ }
+
+#if !defined(_WIN32) && !defined(LWS_WITH_ESP8266)
+ if (wsi->desc.sockfd > context->max_fds) {
+ lwsl_err("fd %d too high (%d)\n", wsi->desc.sockfd, context->max_fds);
return 1;
}
#endif
- if (context->protocols[0].callback(wsi, LWS_CALLBACK_LOCK_POLL,
+ if (wsi->vhost->protocols[0].callback(wsi, LWS_CALLBACK_LOCK_POLL,
wsi->user_space, (void *)&pa, 1))
return -1;
+ lws_same_vh_protocol_remove(wsi);
+
+ /* the guy who is to be deleted's slot index in pt->fds */
+ m = wsi->position_in_fds_table;
+
+#if !defined(LWS_WITH_ESP8266)
lws_libev_io(wsi, LWS_EV_STOP | LWS_EV_READ | LWS_EV_WRITE | LWS_EV_PREPARE_DELETION);
lws_libuv_io(wsi, LWS_EV_STOP | LWS_EV_READ | LWS_EV_WRITE | LWS_EV_PREPARE_DELETION);
lws_pt_lock(pt);
- lwsl_info("%s: wsi=%p, sock=%d, fds pos=%d, end guy pos=%d, endfd=%d\n",
- __func__, wsi, wsi->sock, wsi->position_in_fds_table,
+ lwsl_debug("%s: wsi=%p, sock=%d, fds pos=%d, end guy pos=%d, endfd=%d\n",
+ __func__, wsi, wsi->desc.sockfd, wsi->position_in_fds_table,
pt->fds_count, pt->fds[pt->fds_count].fd);
- /* the guy who is to be deleted's slot index in pt->fds */
- m = wsi->position_in_fds_table;
-
/* have the last guy take up the now vacant slot */
pt->fds[m] = pt->fds[pt->fds_count - 1];
-
+#endif
+ /* this decrements pt->fds_count */
lws_plat_delete_socket_from_fds(context, wsi, m);
-
+#if !defined(LWS_WITH_ESP8266)
+ v = (int) pt->fds[m].fd;
/* end guy's "position in fds table" is now the deletion guy's old one */
- end_wsi = wsi_from_fd(context, pt->fds[pt->fds_count].fd);
- assert(end_wsi);
- end_wsi->position_in_fds_table = m;
+ end_wsi = wsi_from_fd(context, v);
+ if (!end_wsi) {
+ lwsl_err("no wsi found for sock fd %d at pos %d, pt->fds_count=%d\n", (int)pt->fds[m].fd, m, pt->fds_count);
+ assert(0);
+ } else
+ end_wsi->position_in_fds_table = m;
/* deletion guy's lws_lookup entry needs nuking */
- delete_from_fd(context, wsi->sock);
+ delete_from_fd(context, wsi->desc.sockfd);
/* removed wsi has no position any more */
wsi->position_in_fds_table = -1;
/* remove also from external POLL support via protocol 0 */
- if (lws_socket_is_valid(wsi->sock))
- if (context->protocols[0].callback(wsi, LWS_CALLBACK_DEL_POLL_FD,
+ if (lws_socket_is_valid(wsi->desc.sockfd))
+ if (wsi->vhost->protocols[0].callback(wsi, LWS_CALLBACK_DEL_POLL_FD,
wsi->user_space, (void *) &pa, 0))
ret = -1;
#ifndef LWS_NO_SERVER
if (!context->being_destroyed)
/* if this made some room, accept connects on this thread */
if ((unsigned int)pt->fds_count < context->fd_limit_per_thread - 1)
- _lws_change_pollfd(pt->wsi_listening, 0, LWS_POLLIN, &pa1);
+ lws_accept_modulation(pt, 1);
#endif
lws_pt_unlock(pt);
- if (context->protocols[0].callback(wsi, LWS_CALLBACK_UNLOCK_POLL,
+ if (wsi->vhost->protocols[0].callback(wsi, LWS_CALLBACK_UNLOCK_POLL,
wsi->user_space, (void *) &pa, 1))
ret = -1;
-
+#endif
return ret;
}
if (!context)
return 1;
- if (context->protocols[0].callback(wsi, LWS_CALLBACK_LOCK_POLL,
+ if (wsi->vhost->protocols[0].callback(wsi, LWS_CALLBACK_LOCK_POLL,
wsi->user_space, (void *) &pa, 0))
return -1;
lws_pt_lock(pt);
ret = _lws_change_pollfd(wsi, _and, _or, &pa);
lws_pt_unlock(pt);
- if (context->protocols[0].callback(wsi, LWS_CALLBACK_UNLOCK_POLL,
+ if (wsi->vhost->protocols[0].callback(wsi, LWS_CALLBACK_UNLOCK_POLL,
wsi->user_space, (void *) &pa, 0))
ret = -1;
return ret;
}
-
-/**
- * lws_callback_on_writable() - Request a callback when this socket
- * becomes able to be written to without
- * blocking
- *
- * @wsi: Websocket connection instance to get callback for
- */
-
LWS_VISIBLE int
lws_callback_on_writable(struct lws *wsi)
{
+ struct lws_context_per_thread *pt;
#ifdef LWS_USE_HTTP2
struct lws *network_wsi, *wsi2;
int already;
if (wsi->state == LWSS_SHUTDOWN)
return 0;
+ if (wsi->socket_is_permanently_unusable)
+ return 0;
+
+ if (wsi->parent_carries_io) {
+ int n = lws_callback_on_writable(wsi->parent);
+
+ if (n < 0)
+ return n;
+
+ wsi->parent_pending_cb_on_writable = 1;
+ return 1;
+ }
+
+ pt = &wsi->context->pt[(int)wsi->tsi];
+ lws_stats_atomic_bump(wsi->context, pt, LWSSTATS_C_WRITEABLE_CB_REQ, 1);
+#if defined(LWS_WITH_STATS)
+ if (!wsi->active_writable_req_us) {
+ wsi->active_writable_req_us = time_in_microseconds();
+ lws_stats_atomic_bump(wsi->context, pt, LWSSTATS_C_WRITEABLE_CB_EFF_REQ, 1);
+ }
+#endif
+
#ifdef LWS_USE_HTTP2
lwsl_info("%s: %p\n", __func__, wsi);
return 1;
if (wsi->position_in_fds_table < 0) {
- lwsl_err("%s: failed to find socket %d\n", __func__, wsi->sock);
+ lwsl_err("%s: failed to find socket %d\n", __func__, wsi->desc.sockfd);
return -1;
}
return 1;
}
-/**
- * lws_callback_on_writable_all_protocol() - Request a callback for
- * all connections using the given protocol when it
- * becomes possible to write to each socket without
- * blocking in turn.
+/*
+ * stitch protocol choice into the vh protocol linked list
+ * We always insert ourselves at the start of the list
+ *
+ * X <-> B
+ * X <-> pAn <-> pB
*
- * @context: lws_context
- * @protocol: Protocol whose connections will get callbacks
+ * Illegal to attach more than once without detach inbetween
*/
+void
+lws_same_vh_protocol_insert(struct lws *wsi, int n)
+{
+ //lwsl_err("%s: pre insert vhost start wsi %p, that wsi prev == %p\n",
+ // __func__,
+ // wsi->vhost->same_vh_protocol_list[n],
+ // wsi->same_vh_protocol_prev);
+
+ if (wsi->same_vh_protocol_prev || wsi->same_vh_protocol_next) {
+ lws_same_vh_protocol_remove(wsi);
+ lwsl_notice("Attempted to attach wsi twice to same vh prot\n");
+ }
+
+ wsi->same_vh_protocol_prev = /* guy who points to us */
+ &wsi->vhost->same_vh_protocol_list[n];
+ wsi->same_vh_protocol_next = /* old first guy is our next */
+ wsi->vhost->same_vh_protocol_list[n];
+ /* we become the new first guy */
+ wsi->vhost->same_vh_protocol_list[n] = wsi;
+
+ if (wsi->same_vh_protocol_next)
+ /* old first guy points back to us now */
+ wsi->same_vh_protocol_next->same_vh_protocol_prev =
+ &wsi->same_vh_protocol_next;
+}
+
+void
+lws_same_vh_protocol_remove(struct lws *wsi)
+{
+ /*
+ * detach ourselves from vh protocol list if we're on one
+ * A -> B -> C
+ * A -> C , or, B -> C, or A -> B
+ *
+ * OK to call on already-detached wsi
+ */
+ lwsl_info("%s: removing same prot wsi %p\n", __func__, wsi);
+
+ if (wsi->same_vh_protocol_prev) {
+ assert (*(wsi->same_vh_protocol_prev) == wsi);
+ lwsl_info("have prev %p, setting him to our next %p\n",
+ wsi->same_vh_protocol_prev,
+ wsi->same_vh_protocol_next);
+
+ /* guy who pointed to us should point to our next */
+ *(wsi->same_vh_protocol_prev) = wsi->same_vh_protocol_next;
+ }
+
+ /* our next should point back to our prev */
+ if (wsi->same_vh_protocol_next) {
+ wsi->same_vh_protocol_next->same_vh_protocol_prev =
+ wsi->same_vh_protocol_prev;
+ }
+
+ wsi->same_vh_protocol_prev = NULL;
+ wsi->same_vh_protocol_next = NULL;
+}
+
LWS_VISIBLE int
-lws_callback_on_writable_all_protocol(const struct lws_context *context,
+lws_callback_on_writable_all_protocol_vhost(const struct lws_vhost *vhost,
const struct lws_protocols *protocol)
{
- const struct lws_context_per_thread *pt = &context->pt[0];
- unsigned int n, m = context->count_threads;
struct lws *wsi;
- while (m--) {
- for (n = 0; n < pt->fds_count; n++) {
- wsi = wsi_from_fd(context, pt->fds[n].fd);
- if (!wsi)
- continue;
- if (wsi->protocol == protocol)
- lws_callback_on_writable(wsi);
+ if (protocol < vhost->protocols ||
+ protocol >= (vhost->protocols + vhost->count_protocols)) {
+
+ lwsl_err("%s: protocol %p is not from vhost %p (%p - %p)\n",
+ __func__, protocol, vhost->protocols, vhost,
+ (vhost->protocols + vhost->count_protocols));
+
+ return -1;
+ }
+
+ wsi = vhost->same_vh_protocol_list[protocol - vhost->protocols];
+ //lwsl_notice("%s: protocol %p, start wsi %p\n", __func__, protocol, wsi);
+ while (wsi) {
+ //lwsl_notice("%s: protocol %p, this wsi %p (wsi->protocol=%p)\n",
+ // __func__, protocol, wsi, wsi->protocol);
+ assert(wsi->protocol == protocol);
+ assert(*wsi->same_vh_protocol_prev == wsi);
+ if (wsi->same_vh_protocol_next) {
+ // lwsl_err("my next says %p\n", wsi->same_vh_protocol_next);
+ // lwsl_err("my next's prev says %p\n",
+ // wsi->same_vh_protocol_next->same_vh_protocol_prev);
+ assert(wsi->same_vh_protocol_next->same_vh_protocol_prev == &wsi->same_vh_protocol_next);
}
- pt++;
+ //lwsl_notice(" apv: %p\n", wsi);
+ lws_callback_on_writable(wsi);
+ wsi = wsi->same_vh_protocol_next;
+ }
+
+ return 0;
+}
+
+LWS_VISIBLE int
+lws_callback_on_writable_all_protocol(const struct lws_context *context,
+ const struct lws_protocols *protocol)
+{
+ struct lws_vhost *vhost = context->vhost_list;
+ int n;
+
+ while (vhost) {
+ for (n = 0; n < vhost->count_protocols; n++)
+ if (protocol->callback ==
+ vhost->protocols[n].callback &&
+ !strcmp(protocol->name, vhost->protocols[n].name))
+ break;
+ if (n != vhost->count_protocols)
+ lws_callback_on_writable_all_protocol_vhost(
+ vhost, &vhost->protocols[n]);
+
+ vhost = vhost->vhost_next;
}
return 0;
projects / platform / upstream / libwebsockets.git / blobdiff