Commit f44e6e96 authored by Johny Mattsson's avatar Johny Mattsson Committed by Arnim Läuger
Browse files

Fix net module data loss & RTOS task unsafety (#2829)

To avoid races between the lwIP callbacks (lwIP RTOS task) and the Lua
handlers (LVM RTOS task), the data flow and ownership has been simplified
and cleaned up.

lwIP callbacks now have no visibility of the userdata struct. They are
limited to creating small event objects and task_post()ing them over
to the LVM "thread", passing ownership in doing so. The shared identifier
then becomes the struct netconn*.

On the LVM side, we keep a linked list of active userdata objects. This
allows us to retrieve the correct userdata when we get an event with
a netconn pointer. Because this list is only ever used within the LVM
task, no locking is necessary.

The old approach of stashing a userdata pointer into the 'socket' field
on the netconn has been removed entirely, as this was both not
thread/RTOS-task safe, and also interfered with the IDFs internal use
of the socket field (even when using only the netconn layer). As an
added benefit, this removed the need for all the SYS_ARCH_PROTECT()
locking stuff.

The need to track receive events before the corresponding userdata object
has been established has been removed by virtue of not reordering the
"accept" and the "recv" events any more (previously accepts were posted
with medium priority, while the receives where high priority, leading
to the observed reordering and associated headaches).

The workaround for IDF issue 784 has been removed as it is now not needed
and is in fact directly harmful as it results in a double-free. Yay for
getting rid of old workarounds!

DNS resolution code paths were merged for the two instances of "socket"
initiated resolves (connect/dns functions).

Also fixed an instance of using a stack variable for receiving the resolved
IP address, with said variable going out of scope before the DNS resolution
necessarily completed (hello, memory corruption!).

Where possible, moved to use the Lua allocator rather than plain malloc.

Finally, the NodeMCU task posting mechanism got a polish and an adjustment.
Given all the Bad(tm) that tends to happen if something fails task posting,
I went through a couple of iterations on how to avoid that. Alas, the
preferred solution of blocking non-LVM RTOS tasks until a slot is free
turned out to not be viable, as this easily resulted in deadlocks with the
lwIP stack. After much deliberation I settled on increasing the number of
available queue slots for the task_post() mechanism, but in the interest
of user control also now made it user configurable via Kconfig.
parent c0145e03
......@@ -146,6 +146,8 @@ void nodemcu_init(void)
void app_main (void)
{
task_init();
esp_event_queue =
xQueueCreate (CONFIG_SYSTEM_EVENT_QUEUE_SIZE, sizeof (system_event_t));
esp_event_task = task_get_id (handle_esp_event);
......
......@@ -14,9 +14,12 @@
#include "lwip/api.h"
#include "lwip/err.h"
#include "lwip/ip_addr.h"
#include "lwip/dns.h"
#include "lwip/dns.h"
#include "lwip/tcp.h"
#include "lwip/inet.h"
#include "lwip/igmp.h"
#include <esp_log.h>
// Some LWIP macros cause complaints with ptr NULL checks, so shut them off :(
#pragma GCC diagnostic ignored "-Waddress"
......@@ -27,9 +30,7 @@ typedef enum net_type {
TYPE_UDP_SOCKET
} net_type;
typedef const char net_table_name[14];
static const net_table_name NET_TABLES[] = {
static const char *NET_TABLES[] = {
"net.tcpserver",
"net.tcpsocket",
"net.udpsocket"
......@@ -41,6 +42,96 @@ static const net_table_name NET_TABLES[] = {
#define TYPE_TCP TYPE_TCP_CLIENT
#define TYPE_UDP TYPE_UDP_SOCKET
static task_handle_t net_handler;
static task_handle_t dns_handler;
typedef enum {
EVT_SENT, EVT_RECV, EVT_ZEROREAD, EVT_CONNECTED, EVT_ERR
} bounce_event_t;
typedef struct netconn_bounce_event {
struct netconn *netconn;
bounce_event_t event;
err_t err;
uint16_t len;
} netconn_bounce_event_t;
struct lnet_userdata;
typedef struct dns_event {
ip_addr_t resolved_ip;
enum { DNS_STATIC, DNS_SOCKET } invoker;
union {
int cb_ref;
struct lnet_userdata *ud;
};
} dns_event_t;
// --- lwIP callbacks -----------------------------------------------------
// Caution - these run in a different RTOS thread!
static void lnet_netconn_callback(struct netconn *netconn, enum netconn_evt evt, u16_t len)
{
if (!netconn) return;
switch (evt)
{
case NETCONN_EVT_SENDPLUS:
case NETCONN_EVT_RCVPLUS:
case NETCONN_EVT_ERROR: break;
default: return; // we don't care about minus events
}
netconn_bounce_event_t *nbe = malloc(sizeof(netconn_bounce_event_t));
if (!nbe)
{
ESP_LOGE("net", "out of memory - lost event notification!");
return;
}
nbe->netconn = netconn;
nbe->len = len;
nbe->err = ERR_OK;
task_prio_t prio = TASK_PRIORITY_MEDIUM;
switch (evt)
{
case NETCONN_EVT_SENDPLUS:
nbe->event = (len == 0) ? EVT_CONNECTED : EVT_SENT; break;
case NETCONN_EVT_RCVPLUS:
nbe->event = (len == 0) ? EVT_ZEROREAD : EVT_RECV;
prio = TASK_PRIORITY_HIGH;
break;
case NETCONN_EVT_ERROR:
nbe->event = EVT_ERR;
nbe->err = netconn_err(netconn); // is this safe in this thread?
break;
default: break; // impossible
}
if (!task_post(prio, net_handler, (task_param_t)nbe))
ESP_LOGE("net", "out of task post slots - lost data!");
}
// This one may *also* run in the LVM thread
static void net_dns_cb(const char *name, const ip_addr_t *ipaddr, void *callback_arg) {
dns_event_t *ev = (dns_event_t *)callback_arg;
ev->resolved_ip = ipaddr ? *ipaddr : ip_addr_any;
if (!task_post_medium(dns_handler, (task_param_t)ev))
{
ESP_LOGE("net", "out of task post slots - lost DNS response for %s", name);
// And we also leaked the ev here :(
}
}
// --- active list (out of reach of the lwIP thread) ---------------------
typedef struct lnet_userdata {
enum net_type type;
int self_ref;
......@@ -67,35 +158,56 @@ typedef struct lnet_userdata {
int cb_reconnect_ref;
} client;
};
struct lnet_userdata *next;
} lnet_userdata;
static lnet_userdata *active;
// --- Event handling
typedef struct {
enum {
DNSFOUND,
DNSSTATIC,
CONNECTED,
ACCEPT,
RECVDATA,
SENTDATA,
ERR
} event;
union {
lnet_userdata *ud;
int cb_ref;
};
union {
ip_addr_t resolved_ip;
int err;
};
} lnet_event;
static void add_userdata_to_active(lnet_userdata *ud)
{
ud->next = active;
active = ud;
}
static void remove_userdata_from_active(lnet_userdata *ud)
{
if (!ud)
return;
for (lnet_userdata **i = &active; *i; i = &(*i)->next)
{
if (*i == ud)
{
*i = ud->next;
return;
}
}
}
static lnet_userdata *userdata_from_netconn(struct netconn *nc)
{
for (lnet_userdata *i = active; i; i = i->next)
if (i->netconn == nc)
return i;
return NULL;
}
static task_handle_t net_event;
// --- Forward declarations ----------------------------------------------
// --- LWIP errors
static void lsent_cb (lua_State *L, lnet_userdata *ud);
static void lrecv_cb (lua_State *L, lnet_userdata *ud);
static void laccept_cb (lua_State *L, lnet_userdata *ud);
static void lconnected_cb (lua_State *L, lnet_userdata *ud);
static void lerr_cb (lua_State *L, lnet_userdata *ud, err_t err);
static void ldnsstatic_cb (lua_State *L, int cb_ref, ip_addr_t *addr);
static void ldnsfound_cb (lua_State *L, lnet_userdata *ud, ip_addr_t *addr);
// --- Helper functions --------------------------------------------------
int lwip_lua_checkerr (lua_State *L, err_t err) {
switch (err) {
......@@ -119,12 +231,12 @@ int lwip_lua_checkerr (lua_State *L, err_t err) {
}
}
// --- Create
lnet_userdata *net_create( lua_State *L, enum net_type type ) {
const char *mt = NET_TABLES[type];
lnet_userdata *ud = (lnet_userdata *)lua_newuserdata(L, sizeof(lnet_userdata));
if (!ud) return NULL;
luaL_getmetatable(L, mt);
lua_setmetatable(L, -2);
......@@ -151,211 +263,100 @@ lnet_userdata *net_create( lua_State *L, enum net_type type ) {
ud->server.cb_accept_ref = LUA_NOREF;
break;
}
return ud;
}
// --- LWIP callbacks and task_post helpers
static bool post_net_err (lnet_userdata *ud, err_t err) {
lnet_event *ev = (lnet_event *)malloc (sizeof (lnet_event));
if (!ev)
return false;
ev->event = ERR;
ev->ud = ud;
ev->err = err;
if (!task_post_medium (net_event, (task_param_t)ev)) {
free (ev);
return false;
}
return true;
}
static bool post_net_connected (lnet_userdata *ud) {
lnet_event *ev = (lnet_event *)malloc (sizeof (lnet_event));
if (!ev)
return false;
ev->event = CONNECTED;
ev->ud = ud;
if (!task_post_medium (net_event, (task_param_t)ev)) {
free (ev);
return false;
}
return true;
}
add_userdata_to_active(ud);
static bool post_net_dns (lnet_userdata *ud, const char *name, const ip_addr_t *ipaddr)
{
lnet_event *ev = (lnet_event *)malloc (sizeof (lnet_event));
if (!ev || !ipaddr)
return false;
ev->event = DNSFOUND;
ev->ud = ud;
ev->resolved_ip = *ipaddr;
if (!task_post_medium (net_event, (task_param_t)ev)) {
free (ev);
return false;
}
return true;
return ud;
}
static void net_dns_cb(const char *name, const ip_addr_t *ipaddr, void *arg) {
ip_addr_t addr;
if (ipaddr != NULL) addr = *ipaddr;
else addr = ip_addr_any;
lnet_userdata *ud = (lnet_userdata*)arg;
if (!ud) return;
post_net_dns (ud, name, &addr);
}
// --- LVM thread task handlers -----------------------------------------
static bool post_net_recv (lnet_userdata *ud)
void handle_net_event(task_param_t param, task_prio_t prio)
{
lnet_event *ev = (lnet_event *)malloc (sizeof (lnet_event));
if (!ev)
return false;
(void)prio;
netconn_bounce_event_t *pnbe = (netconn_bounce_event_t *)param;
netconn_bounce_event_t nbe = *pnbe;
free(pnbe); // free before we can hit any luaL_error()s
ev->event = RECVDATA;
ev->ud = ud;
lnet_userdata *ud = userdata_from_netconn(nbe.netconn);
if (!ud)
return;
if (!task_post_high (net_event, (task_param_t)ev))
if (nbe.event == EVT_ERR)
{
free (ev);
return false;
}
return true;
}
static bool post_net_sent (lnet_userdata *ud) {
lnet_event *ev = (lnet_event *)malloc (sizeof (lnet_event));
if (!ev)
return false;
ev->event = SENTDATA;
ev->ud = ud;
if (!task_post_medium (net_event, (task_param_t)ev)) {
free (ev);
return false;
}
return true;
}
static bool post_net_accept (lnet_userdata *ud) {
lnet_event *ev = (lnet_event *)malloc (sizeof (lnet_event));
if (!ev)
return false;
ev->event = ACCEPT;
ev->ud = ud;
if (!task_post_medium (net_event, (task_param_t)ev)) {
free (ev);
return false;
}
return true;
}
static void lnet_netconn_callback(struct netconn *netconn, enum netconn_evt evt, u16_t len)
{
if (!netconn) return;
SYS_ARCH_DECL_PROTECT(lev);
SYS_ARCH_PROTECT(lev);
if (netconn->socket < 0) {
if (evt == NETCONN_EVT_RCVPLUS && len > 0) {
// data received before userdata was set up, note receive event
netconn->socket--;
}
SYS_ARCH_UNPROTECT(lev);
lerr_cb(lua_getstate(), ud, nbe.err);
return;
}
SYS_ARCH_UNPROTECT(lev);
lnet_userdata *ud = (lnet_userdata *)netconn->socket;
if (!ud || !ud->netconn) return;
// if a previous event or the user triggered to close the connection then
// skip further event processing, we might run the cbs in Lua task on outdated userdata
if (ud->closing) return;
if (ud->type == TYPE_TCP_CLIENT || ud->type == TYPE_UDP_SOCKET) {
switch (evt) {
case NETCONN_EVT_SENDPLUS:
if (ud->type == TYPE_TCP_CLIENT && ud->client.connecting) {
// connection established, trigger Lua callback
ud->client.connecting = false;
post_net_connected(ud);
} else if (len > 0) {
// we're potentially called back from netconn several times for a single send job
// keep track of them in num_send and postpone the Lua callback until all data is sent
ud->client.num_send -= len;
if (ud->client.num_send == 0) {
// all data sent, trigger Lua callback
post_net_sent(ud);
if (ud->type == TYPE_TCP_CLIENT || ud->type == TYPE_UDP_SOCKET)
{
switch (nbe.event)
{
case EVT_CONNECTED:
if (ud->type == TYPE_TCP_CLIENT && ud->client.connecting)
{
ud->client.connecting = false;
lconnected_cb(lua_getstate(), ud);
}
}
break;
case NETCONN_EVT_ERROR:
post_net_err(ud, netconn_err(ud->netconn));
ud->closing = true;
break;
case NETCONN_EVT_RCVPLUS:
if (len > 0) {
// data received, collect it in Lua callback
post_net_recv(ud);
} else {
// signals closed connection from peer
post_net_err(ud, 0);
break;
case EVT_SENT:
ud->client.num_send -= nbe.len;
if (ud->client.num_send == 0) // all data sent, trigger Lua callback
lsent_cb(lua_getstate(), ud);
break;
case EVT_RECV:
if (nbe.len > 0)
lrecv_cb(lua_getstate(), ud);
break;
case EVT_ZEROREAD: // fall-through
case EVT_ERR:
ud->closing = true;
}
break;
default:
break;
lerr_cb(lua_getstate(), ud, nbe.err);
break;
}
}
else if (ud->type == TYPE_TCP_SERVER)
{
switch (nbe.event)
{
case EVT_ZEROREAD: // new connection
laccept_cb (lua_getstate(), ud);
break;
case EVT_ERR:
ud->closing = true;
lerr_cb(lua_getstate(), ud, nbe.err);
break;
default: break;
}
}
}
} else if (ud->type == TYPE_TCP_SERVER) {
switch (evt) {
case NETCONN_EVT_RCVPLUS:
// new connection available from netconn_listen()
if (ud->netconn &&
ud->server.cb_accept_ref != LUA_NOREF) {
post_net_accept(ud);
}
break;
void handle_dns_event(task_param_t param, task_prio_t prio)
{
(void)prio;
dns_event_t *pev = (dns_event_t *)param;
dns_event_t ev = *pev;
case NETCONN_EVT_ERROR:
post_net_err(ud, netconn_err(ud->netconn));
ud->closing = true;
break;
lua_State *L = lua_getstate();
luaM_free(L, pev); // free before we can hit any luaL_error()s
default:
switch (ev.invoker)
{
case DNS_STATIC:
ldnsstatic_cb(L, ev.cb_ref, &ev.resolved_ip);
break;
}
case DNS_SOCKET:
ldnsfound_cb(L, ev.ud, &ev.resolved_ip);
}
}
// workaround for https://github.com/espressif/esp-idf/issues/784
#include "lwip/priv/api_msg.h"
#define NETCONN_DELETE(conn) \
if (netconn_delete(conn) == ERR_OK) netconn_free(conn);
#define NETCONN_CLOSE(conn) netconn_close_wa(conn)
static err_t netconn_close_wa(struct netconn *conn) {
static err_t netconn_close_and_delete(struct netconn *conn) {
err_t err = netconn_close(conn);
if (err == ERR_OK) {
if (err == ERR_OK)
netconn_delete(conn);
netconn_free(conn);
}
return err;
}
......@@ -363,16 +364,14 @@ static err_t netconn_close_wa(struct netconn *conn) {
// --- Lua API - create
extern int tls_socket_create( lua_State *L );
// Lua: net.createUDPSocket()
int net_createUDPSocket( lua_State *L ) {
static int net_createUDPSocket( lua_State *L ) {
net_create(L, TYPE_UDP_SOCKET);
return 1;
}
// Lua: net.createServer(type, timeout)
int net_createServer( lua_State *L ) {
static int net_createServer( lua_State *L ) {
int type, timeout;
type = luaL_optlong(L, 1, TYPE_TCP);
......@@ -387,7 +386,7 @@ int net_createServer( lua_State *L ) {
}
// Lua: net.createConnection(type, secure)
int net_createConnection( lua_State *L ) {
static int net_createConnection( lua_State *L ) {
int type, secure;
type = luaL_optlong(L, 1, TYPE_TCP);
......@@ -396,11 +395,7 @@ int net_createConnection( lua_State *L ) {
if (type == TYPE_UDP) return net_createUDPSocket( L );
if (type != TYPE_TCP) return luaL_error(L, "invalid type");
if (secure) {
#ifdef CLIENT_SSL_ENABLE
return tls_socket_create( L );
#else
return luaL_error(L, "secure connections not enabled");
#endif
return luaL_error(L, "secure connections not yet supported");
}
net_create(L, TYPE_TCP_CLIENT);
return 1;
......@@ -408,7 +403,7 @@ int net_createConnection( lua_State *L ) {
// --- Get & check userdata
lnet_userdata *net_get_udata_s( lua_State *L, int stack ) {
static lnet_userdata *net_get_udata_s( lua_State *L, int stack ) {
if (!lua_isuserdata(L, stack)) return NULL;
lnet_userdata *ud = (lnet_userdata *)lua_touserdata(L, stack);
switch (ud->type) {
......@@ -424,10 +419,51 @@ lnet_userdata *net_get_udata_s( lua_State *L, int stack ) {
}
#define net_get_udata(L) net_get_udata_s(L, 1)
static int lnet_socket_resolve_dns(lua_State *L, lnet_userdata *ud, const char * domain, bool close_on_err)
{
ud->client.wait_dns ++;
int unref = 0;
if (ud->self_ref == LUA_NOREF) {
unref = 1;
lua_pushvalue(L, 1);
ud->self_ref = luaL_ref(L, LUA_REGISTRYINDEX);
}
dns_event_t *ev = luaM_new(L, dns_event_t);
if (!ev)
return luaL_error (L, "out of memory");
ev->invoker = DNS_SOCKET;
ev->ud = ud;
err_t err = dns_gethostbyname(domain, &ev->resolved_ip, net_dns_cb, ev);
if (err == ERR_OK)
net_dns_cb(domain, &ev->resolved_ip, ev);
else if (err != ERR_INPROGRESS)
{
luaM_free(L, ev);
ud->client.wait_dns --;
if (unref) {
luaL_unref(L, LUA_REGISTRYINDEX, ud->self_ref);
ud->self_ref = LUA_NOREF;
}
if (close_on_err)
{
ud->closing = true;
if (netconn_close_and_delete(ud->netconn) == ERR_OK)
ud->netconn = NULL;
}
return lwip_lua_checkerr(L, err);
}
return 0;
}
// --- Lua API
// Lua: server:listen(port, addr, function(c)), socket:listen(port, addr)
int net_listen( lua_State *L ) {
static int net_listen( lua_State *L ) {
lnet_userdata *ud = net_get_udata(L);
if (!ud || ud->type == TYPE_TCP_CLIENT)
return luaL_error(L, "invalid user data");
......@@ -460,7 +496,6 @@ int net_listen( lua_State *L ) {
ud->netconn = netconn_new_with_callback(NETCONN_TCP, lnet_netconn_callback);
if (!ud->netconn)
return luaL_error(L, "cannot allocate netconn");
ud->netconn->socket = (int)ud;
netconn_set_nonblocking(ud->netconn, 1);
netconn_set_noautorecved(ud->netconn, 1);
......@@ -473,7 +508,6 @@ int net_listen( lua_State *L ) {
ud->netconn = netconn_new_with_callback(NETCONN_UDP, lnet_netconn_callback);
if (!ud->netconn)
return luaL_error(L, "cannot allocate netconn");
ud->netconn->socket = (int)ud;
netconn_set_nonblocking(ud->netconn, 1);
netconn_set_noautorecved(ud->netconn, 1);
......@@ -485,11 +519,11 @@ int net_listen( lua_State *L ) {
ud->closing = true;
switch (ud->type) {
case TYPE_TCP_SERVER:
if (NETCONN_CLOSE(ud->netconn) == ERR_OK)
if (netconn_close_and_delete(ud->netconn) == ERR_OK)
ud->netconn = NULL;
break;
case TYPE_UDP_SOCKET:
if (NETCONN_CLOSE(ud->netconn) == ERR_OK)
if (netconn_close_and_delete(ud->netconn) == ERR_OK)
ud->netconn = NULL;
break;
default: break;
......@@ -504,7 +538,7 @@ int net_listen( lua_State *L ) {
}
// Lua: client:connect(port, addr)
int net_connect( lua_State *L ) {
static int net_connect( lua_State *L ) {
lnet_userdata *ud = net_get_udata(L);
if (!ud || ud->type != TYPE_TCP_CLIENT)
return luaL_error(L, "invalid user data");
......@@ -522,39 +556,15 @@ int net_connect( lua_State *L ) {
ud->netconn = netconn_new_with_callback(NETCONN_TCP, lnet_netconn_callback);
if (!ud->netconn)
return luaL_error(L, "cannot allocate netconn");
ud->netconn->socket = (int)ud;
netconn_set_nonblocking(ud->netconn, 1);
netconn_set_noautorecved(ud->netconn, 1);
ud->port = port;
ip_addr_t addr;
ud->client.wait_dns ++;
int unref = 0;
if (ud->self_ref == LUA_NOREF) {
unref = 1;
lua_pushvalue(L, 1);
ud->self_ref = luaL_ref(L, LUA_REGISTRYINDEX);
}
err_t err = dns_gethostbyname(domain, &addr, net_dns_cb, ud);
if (err == ERR_OK) {
net_dns_cb(domain, &addr, ud);
} else if (err != ERR_INPROGRESS) {
ud->client.wait_dns --;
if (unref) {
luaL_unref(L, LUA_REGISTRYINDEX, ud->self_ref);
ud->self_ref = LUA_NOREF;
}
ud->closing = true;
if (NETCONN_CLOSE(ud->netconn) == ERR_OK)
ud->netconn = NULL;
return lwip_lua_checkerr(L, err);
}
return 0;
return lnet_socket_resolve_dns(L, ud, domain, true);
}
// Lua: client/socket:on(name, callback)
int net_on( lua_State *L ) {
static int net_on( lua_State *L ) {
lnet_userdata *ud = net_get_udata(L);
if (!ud || ud->type == TYPE_TCP_SERVER)
return luaL_error(L, "invalid user data");
......@@ -595,7 +605,7 @@ int net_on( lua_State *L ) {
}
// Lua: client:send(data, function(c)), socket:send(port, ip, data, function(s))
int net_send( lua_State *L ) {
static int net_send( lua_State *L ) {
lnet_userdata *ud = net_get_udata(L);
if (!ud || ud->type == TYPE_TCP_SERVER)
return luaL_error(L, "invalid user data");
......@@ -625,11 +635,10 @@ int net_send( lua_State *L ) {
ud->netconn = netconn_new_with_callback(NETCONN_UDP, lnet_netconn_callback);
if (!ud->netconn)
return luaL_error(L, "cannot allocate netconn");
ud->netconn->socket = (int)ud;
err_t err = netconn_bind(ud->netconn, IP_ADDR_ANY, 0);
if (err != ERR_OK) {
ud->closing = true;
if (NETCONN_CLOSE(ud->netconn) == ERR_OK)
if (netconn_close_and_delete(ud->netconn) == ERR_OK)
ud->netconn = NULL;
return lwip_lua_checkerr(L, err);
}
......@@ -672,7 +681,7 @@ int net_send( lua_State *L ) {
}
// Lua: client:hold()
int net_hold( lua_State *L ) {
static int net_hold( lua_State *L ) {
lnet_userdata *ud = net_get_udata(L);
if (!ud || ud->type != TYPE_TCP_CLIENT)
return luaL_error(L, "invalid user data");
......@@ -688,7 +697,7 @@ int net_hold( lua_State *L ) {
}
// Lua: client:unhold()
int net_unhold( lua_State *L ) {
static int net_unhold( lua_State *L ) {
lnet_userdata *ud = net_get_udata(L);
if (!ud || ud->type != TYPE_TCP_CLIENT)
return luaL_error(L, "invalid user data");
......@@ -705,7 +714,7 @@ int net_unhold( lua_State *L ) {
}
// Lua: client/socket:dns(domain, callback(socket, addr))
int net_dns( lua_State *L ) {
static int net_dns( lua_State *L ) {
lnet_userdata *ud = net_get_udata(L);
if (!ud || ud->type == TYPE_TCP_SERVER)
return luaL_error(L, "invalid user data");
......@@ -720,30 +729,12 @@ int net_dns( lua_State *L ) {
}
if (ud->client.cb_dns_ref == LUA_NOREF)
return luaL_error(L, "no callback specified");
ud->client.wait_dns ++;
int unref = 0;
if (ud->self_ref == LUA_NOREF) {
unref = 1;
lua_pushvalue(L, 1);
ud->self_ref = luaL_ref(L, LUA_REGISTRYINDEX);
}
ip_addr_t addr;
err_t err = dns_gethostbyname(domain, &addr, net_dns_cb, ud);
if (err == ERR_OK) {
net_dns_cb(domain, &addr, ud);
} else if (err != ERR_INPROGRESS) {
ud->client.wait_dns --;
if (unref) {
luaL_unref(L, LUA_REGISTRYINDEX, ud->self_ref);
ud->self_ref = LUA_NOREF;
}
return lwip_lua_checkerr(L, err);
}
return 0;
return lnet_socket_resolve_dns(L, ud, domain, false);
}
// Lua: client:getpeer()
int net_getpeer( lua_State *L ) {
static int net_getpeer( lua_State *L ) {
lnet_userdata *ud = net_get_udata(L);
if (!ud || ud->type != TYPE_TCP_CLIENT)
return luaL_error(L, "invalid user data");
......@@ -770,7 +761,7 @@ int net_getpeer( lua_State *L ) {
}
// Lua: client/server/socket:getaddr()
int net_getaddr( lua_State *L ) {
static int net_getaddr( lua_State *L ) {
lnet_userdata *ud = net_get_udata(L);
if (!ud) return luaL_error(L, "invalid user data");
if (!ud->netconn) {
......@@ -801,7 +792,7 @@ int net_getaddr( lua_State *L ) {
}
// Lua: client/server/socket:close()
int net_close( lua_State *L ) {
static int net_close( lua_State *L ) {
err_t err = ERR_OK;
lnet_userdata *ud = net_get_udata(L);
if (!ud) return luaL_error(L, "invalid user data");
......@@ -811,7 +802,7 @@ int net_close( lua_State *L ) {
case TYPE_TCP_SERVER:
case TYPE_UDP_SOCKET:
ud->closing = true;
err = NETCONN_CLOSE(ud->netconn);
err = netconn_close_and_delete(ud->netconn);
if (err == ERR_OK)
ud->netconn = NULL;
break;
......@@ -830,7 +821,7 @@ int net_close( lua_State *L ) {
return lwip_lua_checkerr(L, err);
}
int net_delete( lua_State *L ) {
static int net_delete( lua_State *L ) {
lnet_userdata *ud = net_get_udata(L);
if (!ud) return luaL_error(L, "no user data");
if (ud->netconn) {
......@@ -839,7 +830,7 @@ int net_delete( lua_State *L ) {
case TYPE_TCP_SERVER:
case TYPE_UDP_SOCKET:
ud->closing = true;
NETCONN_DELETE(ud->netconn);
netconn_delete(ud->netconn);
ud->netconn = NULL;
break;
default: break;
......@@ -869,6 +860,7 @@ int net_delete( lua_State *L ) {
lua_gc(L, LUA_GCSTOP, 0);
luaL_unref(L, LUA_REGISTRYINDEX, ud->self_ref);
ud->self_ref = LUA_NOREF;
remove_userdata_from_active(ud);
lua_gc(L, LUA_GCRESTART, 0);
return 0;
}
......@@ -909,63 +901,54 @@ static int net_multicastJoinLeave( lua_State *L, int join) {
return 0;
}
// Lua: net.multicastJoin(ifip, multicastip)
// if ifip "" or "any" all interfaces are affected
static int net_multicastJoin( lua_State* L ) {
return net_multicastJoinLeave(L,1);
}
// Lua: net.multicastLeave(ifip, multicastip)
// if ifip "" or "any" all interfaces are affected
static int net_multicastLeave( lua_State* L ) {
return net_multicastJoinLeave(L,0);
}
// --- DNS
static void net_dns_static_cb(const char *name, const ip_addr_t *ipaddr, void *callback_arg) {
lnet_event *ev = (lnet_event *)callback_arg;
// --- DNS ---------------------------------------------------------------
ev->resolved_ip = ipaddr ? *ipaddr : ip_addr_any;
if (!task_post_medium (net_event, (task_param_t)ev))
free (ev);
}
// Lua: net.dns.resolve( domain, function(ip) )
static int net_dns_static( lua_State* L ) {
size_t dl;
const char* domain = luaL_checklstring(L, 1, &dl);
if (!domain && dl > 128) {
return luaL_error(L, "wrong domain");
return luaL_error(L, "domain name too long");
}
// Note: this will be free'd using regular free(), so can't luaM_malloc()
lnet_event *ev = (lnet_event *)malloc (sizeof (lnet_event));
luaL_checkanyfunction(L, 2);
lua_settop(L, 2);
dns_event_t *ev = luaM_new(L, dns_event_t);
if (!ev)
return luaL_error (L, "out of memory");
ev->event = DNSSTATIC;
luaL_checkanyfunction(L, 2);
lua_pushvalue(L, 2); // copy argument (func) to the top of stack
ev->cb_ref = luaL_ref(L, LUA_REGISTRYINDEX);
ev->invoker = DNS_STATIC;
err_t err = dns_gethostbyname(
domain, &ev->resolved_ip, net_dns_static_cb, ev);
if (err == ERR_OK) {
net_dns_static_cb(domain, &ev->resolved_ip, ev);
return 0;
} else if (err == ERR_INPROGRESS) {
return 0;
} else {
int e = lwip_lua_checkerr(L, err);
free(ev);
return e;
err_t err = dns_gethostbyname(domain, &ev->resolved_ip, net_dns_cb, ev);
if (err == ERR_OK)
net_dns_cb(domain, &ev->resolved_ip, ev);
else if (err != ERR_INPROGRESS)
{
luaM_free(L, ev);
lwip_lua_checkerr(L, err);
}
return 0;
}
// Lua: s = net.dns.setdnsserver(ip_addr, [index])
static int net_setdnsserver( lua_State* L ) {
size_t l;
......@@ -985,6 +968,7 @@ static int net_setdnsserver( lua_State* L ) {
return 0;
}
// Lua: s = net.dns.getdnsserver([index])
static int net_getdnsserver( lua_State* L ) {
int numdns = luaL_optint(L, 1, 0);
......@@ -1007,7 +991,8 @@ static int net_getdnsserver( lua_State* L ) {
return 1;
}
// --- Lua event dispatch
// --- Lua event dispatch -----------------------------------------------
static void ldnsfound_cb (lua_State *L, lnet_userdata *ud, ip_addr_t *addr) {
if (ud->self_ref != LUA_NOREF && ud->client.cb_dns_ref != LUA_NOREF) {
......@@ -1034,6 +1019,7 @@ static void ldnsfound_cb (lua_State *L, lnet_userdata *ud, ip_addr_t *addr) {
}
}
static void ldnsstatic_cb (lua_State *L, int cb_ref, ip_addr_t *addr) {
if (cb_ref == LUA_NOREF)
return;
......@@ -1051,6 +1037,7 @@ static void ldnsstatic_cb (lua_State *L, int cb_ref, ip_addr_t *addr) {
lua_call(L, 1, 0);
}
static void lconnected_cb (lua_State *L, lnet_userdata *ud) {
if (ud->self_ref != LUA_NOREF && ud->client.cb_connect_ref != LUA_NOREF) {
lua_rawgeti(L, LUA_REGISTRYINDEX, ud->client.cb_connect_ref);
......@@ -1059,43 +1046,6 @@ static void lconnected_cb (lua_State *L, lnet_userdata *ud) {
}
}
static void laccept_cb (lua_State *L, lnet_userdata *ud) {
if (!ud || !ud->netconn) return;
SYS_ARCH_DECL_PROTECT(lev);
lua_rawgeti(L, LUA_REGISTRYINDEX, ud->server.cb_accept_ref);
lnet_userdata *nud = net_create(L, TYPE_TCP_CLIENT);
lua_pushvalue(L, -1);
nud->self_ref = luaL_ref(L, LUA_REGISTRYINDEX);
int recvevent = 0;
struct netconn *newconn;
err_t err = netconn_accept(ud->netconn, &newconn);
if (err == ERR_OK) {
nud->netconn = newconn;
SYS_ARCH_PROTECT(lev);
// take buffered receive events
recvevent = (int)(-1 - newconn->socket);
nud->netconn->socket = (int)nud;
SYS_ARCH_UNPROTECT(lev);
netconn_set_nonblocking(nud->netconn, 1);
netconn_set_noautorecved(nud->netconn, 1);
nud->netconn->pcb.tcp->so_options |= SOF_KEEPALIVE;
nud->netconn->pcb.tcp->keep_idle = ud->server.timeout * 1000;
nud->netconn->pcb.tcp->keep_cnt = 1;
} else
luaL_error(L, "cannot accept new server socket connection");
lua_call(L, 1, 0);
while (recvevent-- > 0) {
// kick receive callback in case of pending events
post_net_recv(nud);
}
}
static void lrecv_cb (lua_State *L, lnet_userdata *ud) {
if (!ud || !ud->netconn) return;
......@@ -1109,6 +1059,7 @@ static void lrecv_cb (lua_State *L, lnet_userdata *ud) {
lwip_lua_checkerr(L, err);
return;
}
netbuf_first(p);
do {
netbuf_data(p, (void **)&payload, &len);
......@@ -1144,6 +1095,31 @@ static void lrecv_cb (lua_State *L, lnet_userdata *ud) {
}
}
static void laccept_cb (lua_State *L, lnet_userdata *ud) {
if (!ud || !ud->netconn) return;
lua_rawgeti(L, LUA_REGISTRYINDEX, ud->server.cb_accept_ref);
lnet_userdata *nud = net_create(L, TYPE_TCP_CLIENT);
lua_pushvalue(L, -1);
nud->self_ref = luaL_ref(L, LUA_REGISTRYINDEX);
struct netconn *newconn;
err_t err = netconn_accept(ud->netconn, &newconn);
if (err == ERR_OK) {
nud->netconn = newconn;
netconn_set_nonblocking(nud->netconn, 1);
netconn_set_noautorecved(nud->netconn, 1);
nud->netconn->pcb.tcp->so_options |= SOF_KEEPALIVE;
nud->netconn->pcb.tcp->keep_idle = ud->server.timeout * 1000;
nud->netconn->pcb.tcp->keep_cnt = 1;
} else
luaL_error(L, "cannot accept new server socket connection");
lua_call(L, 1, 0);
}
static void lsent_cb (lua_State *L, lnet_userdata *ud) {
if (ud->client.cb_sent_ref != LUA_NOREF) {
lua_rawgeti(L, LUA_REGISTRYINDEX, ud->client.cb_sent_ref);
......@@ -1152,6 +1128,7 @@ static void lsent_cb (lua_State *L, lnet_userdata *ud) {
}
}
static void lerr_cb (lua_State *L, lnet_userdata *ud, err_t err)
{
if (!ud || !ud->netconn) return;
......@@ -1174,30 +1151,9 @@ static void lerr_cb (lua_State *L, lnet_userdata *ud, err_t err)
}
}
static void handle_net_event (task_param_t param, task_prio_t prio)
{
lnet_event *ev = (lnet_event *)param;
(void)prio;
lua_State *L = lua_getstate();
switch (ev->event)
{
case DNSFOUND: ldnsfound_cb (L, ev->ud, &ev->resolved_ip); break;
case DNSSTATIC: ldnsstatic_cb (L, ev->cb_ref, &ev->resolved_ip); break;
case CONNECTED: lconnected_cb (L, ev->ud); break;
case ACCEPT: laccept_cb (L, ev->ud); break;
case RECVDATA: lrecv_cb (L, ev->ud); break;
case SENTDATA: lsent_cb (L, ev->ud); break;
case ERR: lerr_cb (L, ev->ud, ev->err); break;
}
free (ev);
}
// --- Tables
extern const LUA_REG_TYPE tls_cert_map[];
// Module function map
static const LUA_REG_TYPE net_tcpserver_map[] = {
{ LSTRKEY( "listen" ), LFUNCVAL( net_listen ) },
......@@ -1249,15 +1205,13 @@ static const LUA_REG_TYPE net_map[] = {
{ LSTRKEY( "multicastJoin"), LFUNCVAL( net_multicastJoin ) },
{ LSTRKEY( "multicastLeave"), LFUNCVAL( net_multicastLeave ) },
{ LSTRKEY( "dns" ), LROVAL( net_dns_map ) },
#ifdef CLIENT_SSL_ENABLE
{ LSTRKEY( "cert" ), LROVAL( tls_cert_map ) },
#endif
{ LSTRKEY( "TCP" ), LNUMVAL( TYPE_TCP ) },
{ LSTRKEY( "UDP" ), LNUMVAL( TYPE_UDP ) },
{ LSTRKEY( "__metatable" ), LROVAL( net_map ) },
{ LNILKEY, LNILVAL }
};
int luaopen_net( lua_State *L ) {
igmp_init();
......@@ -1265,7 +1219,8 @@ int luaopen_net( lua_State *L ) {
luaL_rometatable(L, NET_TABLE_TCP_CLIENT, (void *)net_tcpsocket_map);
luaL_rometatable(L, NET_TABLE_UDP_SOCKET, (void *)net_udpsocket_map);
net_event = task_get_id (handle_net_event);
net_handler = task_get_id(handle_net_event);
dns_handler = task_get_id(handle_dns_event);
return 0;
}
......
menu "NodeMCU task slot configuration"
config NODEMCU_TASK_SLOT_MEMORY
int "Task slot buffer size"
default 2000
range 80 16000
help
NodeMCU uses a fixed size RTOS queue for messaging between internal
LVM tasks as well as from other RTOS tasks. If this queue is too
small, events and data will go missing. On the other hand, if the
queue is too big, some memory will go unused.
The default value is chosen to be on the safe side for most use
cases. Lowering this value will yield more available RAM for use
in Lua, but at the increased risk of data loss. Conversely,
increasing this value can help resolve aforementioned data loss
issues, if encountered.
The assigned memory size here gets partitioned to the different
task priorities; some rounding down may take place as a result.
endmenu
......@@ -29,9 +29,11 @@ bool task_post(task_prio_t priority, task_handle_t handle, task_param_t param);
typedef void (*task_callback_t)(task_param_t param, task_prio_t prio);
bool task_init_handler(task_prio_t priority, uint8 qlen);
task_handle_t task_get_id(task_callback_t t);
/* Init, must be called before any posting happens */
void task_init (void);
/* RTOS loop to pump task messages until infinity */
void task_pump_messages (void);
......
......@@ -13,7 +13,6 @@
#define TASK_HANDLE_MASK 0xFFF80000
#define TASK_HANDLE_UNMASK (~TASK_HANDLE_MASK)
#define TASK_HANDLE_ALLOCATION_BRICK 4 // must be a power of 2
#define TASK_DEFAULT_QUEUE_LEN 8
#define CHECK(p,v,msg) if (!(p)) { NODE_DBG ( msg ); return (v); }
......@@ -40,33 +39,7 @@ static task_callback_t *task_func;
static int task_count;
/*
* Initialise the task handle callback for a given priority. This doesn't need
* to be called explicitly as the get_id function will call this lazily.
*/
bool task_init_handler(task_prio_t priority, uint8 qlen) {
if (priority >= TASK_PRIORITY_COUNT)
return false;
if (task_Q[priority] == NULL)
{
task_Q[priority] = xQueueCreate (qlen, sizeof (task_event_t));
return task_Q[priority] != NULL;
}
else
return false;
}
task_handle_t task_get_id(task_callback_t t) {
/* Initialise any uninitialised Qs with the default Q len */
for (task_prio_t p = TASK_PRIORITY_LOW; p != TASK_PRIORITY_COUNT; ++p)
{
if (!task_Q[p]) {
CHECK(task_init_handler( p, TASK_DEFAULT_QUEUE_LEN ), 0, "Task initialisation failed");
}
}
if ( (task_count & (TASK_HANDLE_ALLOCATION_BRICK - 1)) == 0 ) {
/* With a brick size of 4 this branch is taken at 0, 4, 8 ... and the new size is +4 */
task_func =(task_callback_t *)realloc(
......@@ -85,15 +58,13 @@ task_handle_t task_get_id(task_callback_t t) {
bool task_post (task_prio_t priority, task_handle_t handle, task_param_t param)
{
if (priority >= TASK_PRIORITY_COUNT ||
!task_Q[priority] ||
(handle & TASK_HANDLE_MASK) != TASK_HANDLE_MONIKER)
return false;
task_event_t ev = { handle, param };
bool res = pdPASS == xQueueSendToBackFromISR (task_Q[priority], &ev, NULL);
if (pending) /* only need to raise semaphore if it's been initialised */
xSemaphoreGiveFromISR (pending, NULL);
xSemaphoreGiveFromISR (pending, NULL);
return res;
}
......@@ -129,9 +100,30 @@ static void dispatch (task_event_t *e, uint8_t prio) {
}
void task_init (void)
{
pending = xSemaphoreCreateBinary ();
// Due to the nature of the RTOS, if we ever fail to do a task_post, we're
// up the proverbial creek without a paddle. Each queue slot only costs us
// 8 bytes, so it's a worthwhile trade-off to reserve a bit of RAM for this
// purpose. Trying to work around the issue in the places which post ends
// up being *at least* as bad, so better take the hit where the benefit
// is shared. That said, we let the user have the final say.
const size_t q_mem = CONFIG_NODEMCU_TASK_SLOT_MEMORY;
// Rather than blindly sizing everything the same, we try to be a little
// bit aware of the typical uses of the queues.
const size_t slice = q_mem / (10 * sizeof(task_event_t));
static const int qlens[] = { 1 * slice, 5 * slice, 4 * slice };
for (task_prio_t p = TASK_PRIORITY_LOW; p != TASK_PRIORITY_COUNT; ++p)
task_Q[p] = xQueueCreate (qlens[p], sizeof (task_event_t));
}
void task_pump_messages (void)
{
vSemaphoreCreateBinary (pending);
for (;;)
{
task_event_t ev;
......
Markdown is supported
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment