stv0g/libwebsockets
1
0
Fork 0
Code Issues Pull requests Releases Wiki Activity

add locking callback for fds

Browse source 
This adds two new callbacks in protocols[0] that are optional for allowing limited thread
access to libwebsockets, LWS_CALLBACK_LOCK_POLL and LWS_CALLBACK_UNLOCK_POLL.

If you use them, they protect internal and external poll list changes, but if you want to use
external thread access to libwebsocket_callback_on_writable() you have to implement your
locking here even if you don't use external poll support.

If you will use another thread for this, take a lot of care about managing your list of
live wsi by doing it from ESTABLISHED and CLOSED callbacks (with your own locking).

Signed-off-by: Andy Green <andy.green@linaro.org>
This commit is contained in:
Andy Green 2013-12-18 09:48:26 +08:00
parent 5b34c975ae
commit 7a1327977a
6 changed files with 119 additions and 1 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

15
changelog
View file

@ -26,6 +26,21 @@ also be accepted on an SSL listening port. It's disabled unless you enable
it explicitly.
Two new callbacks are added in protocols[0] that are optional for allowing
limited thread access to libwebsockets, LWS_CALLBACK_LOCK_POLL and
LWS_CALLBACK_UNLOCK_POLL.
If you use them, they protect internal and external poll list changes, but if
you want to use external thread access to libwebsocket_callback_on_writable()
you have to implement your locking here even if you don't use external
poll support.
If you will use another thread for this, take a lot of care about managing
your list of live wsi by doing it from ESTABLISHED and CLOSED callbacks
(with your own locking).
User api changes
----------------

8
lib/client.c
View file

@ -119,6 +119,10 @@ int lws_client_socket_service(struct libwebsocket_context *context,
* happening at a time when there's no real connection yet
*/
context->protocols[0].callback(context, wsi,
LWS_CALLBACK_LOCK_POLL,
wsi->user_space, (void *)(long)wsi->sock, 0);
pollfd->events &= ~POLLOUT;
/* external POLL support via protocol 0 */
@ -126,6 +130,10 @@ int lws_client_socket_service(struct libwebsocket_context *context,
LWS_CALLBACK_CLEAR_MODE_POLL_FD,
wsi->user_space, (void *)(long)wsi->sock, POLLOUT);
context->protocols[0].callback(context, wsi,
LWS_CALLBACK_UNLOCK_POLL,
wsi->user_space, (void *)(long)wsi->sock, 0);
/* we can retry this... just cook the SSL BIO the first time */
if (wsi->use_ssl && !wsi->ssl) {

47
lib/libwebsockets.c
View file

@ -118,6 +118,10 @@ insert_wsi_socket_into_fds(struct libwebsocket_context *context,
lwsl_info("insert_wsi_socket_into_fds: wsi=%p, sock=%d, fds pos=%d\n",
wsi, wsi->sock, context->fds_count);
context->protocols[0].callback(context, wsi,
LWS_CALLBACK_LOCK_POLL,
wsi->user_space, (void *)(long)wsi->sock, 0);
context->lws_lookup[wsi->sock] = wsi;
wsi->position_in_fds_table = context->fds_count;
context->fds[context->fds_count].fd = wsi->sock;
@ -129,6 +133,10 @@ insert_wsi_socket_into_fds(struct libwebsocket_context *context,
LWS_CALLBACK_ADD_POLL_FD,
wsi->user_space, (void *)(long)wsi->sock, POLLIN);
context->protocols[0].callback(context, wsi,
LWS_CALLBACK_UNLOCK_POLL,
wsi->user_space, (void *)(long)wsi->sock, 0);
return 0;
}
@ -138,8 +146,12 @@ remove_wsi_socket_from_fds(struct libwebsocket_context *context,
{
int m;
if (!--context->fds_count)
if (!--context->fds_count) {
context->protocols[0].callback(context, wsi,
LWS_CALLBACK_LOCK_POLL,
wsi->user_space, (void *)(long)wsi->sock, 0);
goto do_ext;
}
if (wsi->sock > context->max_fds) {
lwsl_err("Socket fd %d too high (%d)\n",
@ -150,6 +162,10 @@ remove_wsi_socket_from_fds(struct libwebsocket_context *context,
lwsl_info("remove_wsi_socket_from_fds: wsi=%p, sock=%d, fds pos=%d\n",
wsi, wsi->sock, wsi->position_in_fds_table);
context->protocols[0].callback(context, wsi,
LWS_CALLBACK_LOCK_POLL,
wsi->user_space, (void *)(long)wsi->sock, 0);
m = wsi->position_in_fds_table; /* replace the contents for this */
/* have the last guy take up the vacant slot */
@ -173,6 +189,10 @@ do_ext:
LWS_CALLBACK_DEL_POLL_FD, wsi->user_space,
(void *)(long)wsi->sock, 0);
context->protocols[0].callback(context, wsi,
LWS_CALLBACK_UNLOCK_POLL,
wsi->user_space, (void *)(long)wsi->sock, 0);
return 0;
}
@ -789,12 +809,21 @@ user_service:
/* one shot */
if (pollfd) {
context->protocols[0].callback(context, wsi,
LWS_CALLBACK_LOCK_POLL,
wsi->user_space, (void *)(long)wsi->sock, 0);
pollfd->events &= ~POLLOUT;
/* external POLL support via protocol 0 */
context->protocols[0].callback(context, wsi,
LWS_CALLBACK_CLEAR_MODE_POLL_FD,
wsi->user_space, (void *)(long)wsi->sock, POLLOUT);
context->protocols[0].callback(context, wsi,
LWS_CALLBACK_UNLOCK_POLL,
wsi->user_space, (void *)(long)wsi->sock, 0);
}
#ifndef LWS_NO_EXTENSIONS
notify_action:
@ -1427,6 +1456,10 @@ libwebsocket_callback_on_writable(struct libwebsocket_context *context,
return -1;
}
context->protocols[0].callback(context, wsi,
LWS_CALLBACK_LOCK_POLL,
wsi->user_space, (void *)(long)wsi->sock, 0);
context->fds[wsi->position_in_fds_table].events |= POLLOUT;
/* external POLL support via protocol 0 */
@ -1434,6 +1467,10 @@ libwebsocket_callback_on_writable(struct libwebsocket_context *context,
LWS_CALLBACK_SET_MODE_POLL_FD,
wsi->user_space, (void *)(long)wsi->sock, POLLOUT);
context->protocols[0].callback(context, wsi,
LWS_CALLBACK_UNLOCK_POLL,
wsi->user_space, (void *)(long)wsi->sock, 0);
return 1;
}
@ -1578,6 +1615,10 @@ _libwebsocket_rx_flow_control(struct libwebsocket *wsi)
/* adjust the pollfd for this wsi */
context->protocols[0].callback(context, wsi,
LWS_CALLBACK_LOCK_POLL,
wsi->user_space, (void *)(long)wsi->sock, 0);
if (wsi->u.ws.rxflow_change_to & LWS_RXFLOW_ALLOW)
context->fds[wsi->position_in_fds_table].events |= POLLIN;
else
@ -1594,6 +1635,10 @@ _libwebsocket_rx_flow_control(struct libwebsocket *wsi)
LWS_CALLBACK_CLEAR_MODE_POLL_FD,
wsi->user_space, (void *)(long)wsi->sock, POLLIN);
context->protocols[0].callback(context, wsi,
LWS_CALLBACK_UNLOCK_POLL,
wsi->user_space, (void *)(long)wsi->sock, 0);
return 1;
}
#endif

2
lib/libwebsockets.h
View file

@ -174,6 +174,8 @@ enum libwebsocket_callback_reasons {
LWS_CALLBACK_DEL_POLL_FD,
LWS_CALLBACK_SET_MODE_POLL_FD,
LWS_CALLBACK_CLEAR_MODE_POLL_FD,
LWS_CALLBACK_LOCK_POLL,
LWS_CALLBACK_UNLOCK_POLL,
};
#ifndef LWS_NO_EXTENSIONS

34
lib/server.c
View file

@ -212,8 +212,22 @@ int lws_server_socket_service(struct libwebsocket_context *context,
break;
/* one shot */
context->protocols[0].callback(context, wsi,
LWS_CALLBACK_LOCK_POLL,
wsi->user_space, (void *)(long)wsi->sock, 0);
pollfd->events &= ~POLLOUT;
/* external POLL support via protocol 0 */
context->protocols[0].callback(context, wsi,
LWS_CALLBACK_CLEAR_MODE_POLL_FD,
wsi->user_space, (void *)(long)wsi->sock, POLLOUT);
context->protocols[0].callback(context, wsi,
LWS_CALLBACK_UNLOCK_POLL,
wsi->user_space, (void *)(long)wsi->sock, 0);
if (wsi->state != WSI_STATE_HTTP_ISSUING_FILE) {
n = user_callback_handle_rxflow(
wsi->protocol->callback,
@ -346,6 +360,10 @@ int lws_server_socket_service(struct libwebsocket_context *context,
case LWS_CONNMODE_SSL_ACK_PENDING:
context->protocols[0].callback(context, wsi,
LWS_CALLBACK_LOCK_POLL,
wsi->user_space, (void *)(long)wsi->sock, 0);
pollfd->events &= ~POLLOUT;
/* external POLL support via protocol 0 */
@ -353,6 +371,10 @@ int lws_server_socket_service(struct libwebsocket_context *context,
LWS_CALLBACK_CLEAR_MODE_POLL_FD,
wsi->user_space, (void *)(long)wsi->sock, POLLOUT);
context->protocols[0].callback(context, wsi,
LWS_CALLBACK_UNLOCK_POLL,
wsi->user_space, (void *)(long)wsi->sock, 0);
lws_latency_pre(context, wsi);
n = recv(wsi->sock, context->service_buffer,
@ -395,6 +417,9 @@ int lws_server_socket_service(struct libwebsocket_context *context,
m, ERR_error_string(m, NULL));
if (m == SSL_ERROR_WANT_READ) {
context->protocols[0].callback(context, wsi,
LWS_CALLBACK_LOCK_POLL,
wsi->user_space, (void *)(long)wsi->sock, 0);
context->fds[
wsi->position_in_fds_table].events |= POLLIN;
@ -403,10 +428,16 @@ int lws_server_socket_service(struct libwebsocket_context *context,
LWS_CALLBACK_SET_MODE_POLL_FD,
wsi->user_space,
(void *)(long)wsi->sock, POLLIN);
context->protocols[0].callback(context, wsi,
LWS_CALLBACK_UNLOCK_POLL,
wsi->user_space, (void *)(long)wsi->sock, 0);
lwsl_info("SSL_ERROR_WANT_READ\n");
break;
}
if (m == SSL_ERROR_WANT_WRITE) {
context->protocols[0].callback(context, wsi,
LWS_CALLBACK_LOCK_POLL,
wsi->user_space, (void *)(long)wsi->sock, 0);
context->fds[
wsi->position_in_fds_table].events |= POLLOUT;
/* external POLL support via protocol 0 */
@ -414,6 +445,9 @@ int lws_server_socket_service(struct libwebsocket_context *context,
LWS_CALLBACK_SET_MODE_POLL_FD,
wsi->user_space,
(void *)(long)wsi->sock, POLLOUT);
context->protocols[0].callback(context, wsi,
LWS_CALLBACK_UNLOCK_POLL,
wsi->user_space, (void *)(long)wsi->sock, 0);
break;
}
lwsl_debug("SSL_accept failed skt %u: %s\n",

14
test-server/test-server.c
View file

@ -429,6 +429,20 @@ bail:
* protocol 0 callback
*/
case LWS_CALLBACK_LOCK_POLL:
/*
* lock mutex to protect pollfd state
* called before any other POLL related callback
*/
break;
case LWS_CALLBACK_UNLOCK_POLL:
/*
* unlock mutex to protect pollfd state when
* called after any other POLL related callback
*/
break;
case LWS_CALLBACK_ADD_POLL_FD:
if (count_pollfds >= max_poll_elements) {