Fixed lwIP driver thread model violation (tcp_close() was being called by application thread)

include/libzt.h

@@ -213,11 +213,6 @@ struct zts_ifreq {
 #define ZT_HOME_PATH_MAX_LEN               128
 #define ZT_MAC_ADDRSTRLEN                  18
 
-#define ZT_SOCK_STATE_NONE                 100
-#define ZT_SOCK_STATE_UNHANDLED_CONNECTED  101
-#define ZT_SOCK_STATE_CONNECTED            102
-#define ZT_SOCK_STATE_LISTENING            103
-
 #define ZT_ERR_OK                          0
 #define ZT_ERR_GENERAL_FAILURE             -88
 
@@ -594,7 +589,7 @@ namespace ZeroTier
 	extern ZeroTier::lwIP *lwipstack;
 
 	class VirtualTap;
-	struct VirtualSocket;
+	class VirtualSocket;
 	struct InetAddress;
 }
 

src/VirtualSocket.hpp

@@ -45,6 +45,13 @@
 #include "VirtualTap.hpp"
 #include "RingBuffer.hpp"
 
+#define VS_OK                         100
+#define VS_SHOULD_STOP                101
+#define VS_STOPPED                    102
+#define VS_STATE_UNHANDLED_CONNECTED  103
+#define VS_STATE_CONNECTED            104
+#define VS_STATE_LISTENING            105
+
 namespace ZeroTier {
 
 	class VirtualTap;
@@ -56,13 +63,29 @@ namespace ZeroTier {
 	 * function of this object, however I'd like to discourage this since this
 	 * object also handles non-connection-based traffic as well.
 	 */
-	struct VirtualSocket
+	class VirtualSocket
 	{
+	private:
+		int _state = VS_OK;
+	public:
 		RingBuffer<unsigned char> *TXbuf;
 		RingBuffer<unsigned char> *RXbuf;
-		Mutex _tx_m, _rx_m;
+		Mutex _tx_m, _rx_m, _op_m;
 		PhySocket *sock = NULL;
 
+		// State control
+		void set_state(int state) {
+			// states may be set by application or by stack callbacks, thus this must be guarded
+			_op_m.lock();
+			_state = state; 
+			//DEBUG_EXTRA("SET STATE = %d (vs=%p)", _state, this);
+			_op_m.unlock();
+		}
+		int get_state() {
+			//DEBUG_EXTRA("GET STATE = %d (vs=%p)", _state, this);
+			return _state; 
+		}
+
 #if defined(STACK_PICO)
 		struct pico_socket *picosock = NULL;
 #endif
@@ -88,7 +111,6 @@ namespace ZeroTier {
 		int socket_family = 0;
 		int socket_type = 0;
 		int protocol = 0;
-		int state = 0; // See libzt.h for (ZT_SOCK_STATE_*)
 		int app_fd = 0; // used by app for I/O
 		int sdk_fd = 0; // used by lib for I/O
 

src/VirtualTap.cpp

@@ -719,7 +719,7 @@ namespace ZeroTier {
 		Mutex::Lock _l(_tcpconns_m);
 		std::time_t current_ts = std::time(nullptr);
 		if (current_ts > last_housekeeping_ts + ZT_HOUSEKEEPING_INTERVAL) {
-			DEBUG_EXTRA();
+			// DEBUG_EXTRA();
 			// update managed routes (add/del from network stacks)
 			ZeroTier::OneService *service = ((ZeroTier::OneService *)zt1ServiceRef);
 			if (service) {

src/lwIP.cpp

@@ -643,51 +643,53 @@ namespace ZeroTier
 
 	int lwIP::lwip_Close(VirtualSocket *vs)
 	{
+		// requests to close non-LISTEN PCBs are handled lwip_cb_poll()
+		int err = -1;
 		if (vs == NULL) {
 			DEBUG_ERROR("invalid vs");
 			handle_general_failure();
 			return -1;
 		}
-		DEBUG_EXTRA("fd=%d, vs=%p", vs->app_fd, vs);
-		int err = 0;
-		errno = 0;
-		if (vs->socket_type == SOCK_DGRAM) {
-			udp_remove((struct udp_pcb*)vs->pcb);
-		}
 		if (vs->socket_type == SOCK_STREAM) {
-			if (vs->pcb) {
+			struct tcp_pcb *tpcb = (struct tcp_pcb*)(vs->pcb);
-				struct tcp_pcb* tpcb = (struct tcp_pcb*)vs->pcb;
+			if (tpcb == NULL) {
-				if (tpcb->state == CLOSED) {
+				DEBUG_ERROR("invalid pcb");
-					DEBUG_EXTRA("pcb is in CLOSED state");
+				handle_general_failure();
-					// calling tcp_close() here would be redundant
+				return -1;
-					return 0;
+			}
-				}
+			// should be safe to tcp_close() from application thread IF PCB is in LISTENING state (I think)
-				if (tpcb->state == CLOSE_WAIT) {
+			if (tpcb->state == LISTEN) {
-					DEBUG_EXTRA("pcb is in CLOSE_WAIT state");
+				DEBUG_EXTRA("PCB is in LISTEN, calling tcp_close() from application thread.");
-					// calling tcp_close() here would be redundant
+				tcp_accept(tpcb, NULL);
-				}
-				if (tpcb->state > TIME_WAIT) {
-					DEBUG_ERROR("warning, pcb=%p is in an invalid state=%d", vs->pcb, tpcb->state);
-					handle_general_failure();
-					err = -1;
-				}
-				// unregister callbacks for this PCB
-				tcp_arg(tpcb, NULL);
-				if (tpcb->state == LISTEN) {
-					tcp_accept(tpcb, NULL);
-				}
-				else {
-					tcp_recv(tpcb, NULL);
-					tcp_sent(tpcb, NULL);
-					tcp_poll(tpcb, NULL, 0);
-					tcp_err(tpcb,  NULL);
-				}
 				if ((err = tcp_close(tpcb)) < 0) {
 					DEBUG_ERROR("error while calling tcp_close, fd=%d, vs=%p, pcb=%p", vs->app_fd, vs, vs->pcb);
 					errno = lwip_err_to_errno(err);
 					err = -1;
 				}
+				return ERR_OK;
 			}
+			// handle junk state values
+			if (tpcb->state > TIME_WAIT) {
+				DEBUG_EXTRA("invalid TCP pcb state, already closed, report ERR_OK");
+				return ERR_OK;
+			}
+			else {
+				// place a request for the stack to close this VirtualSocket's PCB
+				vs->set_state(VS_SHOULD_STOP);
+				// wait for indication of success, this will block if the PCB can't close
+				while (true) {
+					sleep(1);
+					nanosleep((const struct timespec[]) {{0, (ZT_API_CHECK_INTERVAL * 1000000)}}, NULL);
+					DEBUG_EXTRA("checking closure state... pcb->state=%d", tpcb->state);
+					if (vs->get_state() == VS_STOPPED || tpcb->state == CLOSED) {
+						return ERR_OK;
+					}
+				}
+			}	
+		}
+		if (vs->socket_type == SOCK_DGRAM) {
+			// place a request for the stack to close this VirtualSocket's PCB
+			vs->set_state(VS_SHOULD_STOP);
 		}
 		return err;
 	}
@@ -740,6 +742,7 @@ namespace ZeroTier
 		struct pbuf* q = p;
 		if (p == NULL) {
 			DEBUG_INFO("p=0x0 for pcb=%p, vs->pcb=%p, this indicates a closure. No need to call tcp_close()", PCB, vs->pcb);
+			vs->set_state(VS_SHOULD_STOP);
 			return ERR_ABRT;
 		}
 		vs->tap->_tcpconns_m.lock();
@@ -876,7 +879,6 @@ namespace ZeroTier
 			// payload
 			memcpy(udp_msg_buf + sizeof(int32_t) + sizeof(struct sockaddr_storage), &udp_payload_buf, tot_len); 
 			if ((w = write(vs->sdk_fd, udp_msg_buf, msg_tot_len)) < 0) {
-				perror("write");
 				DEBUG_ERROR("write(fd=%d)=%d, errno=%d", vs->sdk_fd, w, errno);
 			}
 		}
@@ -933,7 +935,7 @@ namespace ZeroTier
 		}
 		// add to unhandled connection set for zts_connect to pick up on
 		vs->tap->_tcpconns_m.lock();
-		vs->state = ZT_SOCK_STATE_UNHANDLED_CONNECTED;
+		vs->set_state(VS_STATE_UNHANDLED_CONNECTED);
 		vs->tap->_VirtualSockets.push_back(vs);
 		vs->tap->_tcpconns_m.unlock();
 		return ERR_OK;
@@ -950,6 +952,58 @@ namespace ZeroTier
 		if (vs->socket_type == SOCK_DGRAM) {
 			DEBUG_INFO("fd=%d, vs=%p, pcb=%p", vs->app_fd, vs, PCB, vs->pcb);
 		}
+
+
+		// Handle PCB closure requests (set in lwip_Close())
+		if (vs->get_state() == VS_SHOULD_STOP) {
+			DEBUG_EXTRA("closing pcb=%p, fd=%d, vs=%p", PCB, vs->app_fd, vs);
+			int err = 0;
+			errno = 0;
+			if (vs->socket_type == SOCK_DGRAM) {
+				udp_remove((struct udp_pcb*)vs->pcb);
+			}
+			if (vs->socket_type == SOCK_STREAM) {
+				if (vs->pcb) {
+					struct tcp_pcb* tpcb = (struct tcp_pcb*)vs->pcb;
+					if (tpcb->state == CLOSED) {
+						DEBUG_EXTRA("pcb is in CLOSED state");
+						// calling tcp_close() here would be redundant
+						return 0;
+					}
+					//if (tpcb->state == CLOSE_WAIT) {
+					//	DEBUG_EXTRA("pcb is in CLOSE_WAIT state");
+					//	// calling tcp_close() here would be redundant
+					//}
+					if (tpcb->state > TIME_WAIT) {
+						DEBUG_ERROR("warning, pcb=%p is in an invalid state=%d", vs->pcb, tpcb->state);
+						handle_general_failure();
+						err = -1;
+					}
+					// unregister callbacks for this PCB
+					tcp_arg(tpcb, NULL);
+					if (tpcb->state == LISTEN) {
+						tcp_accept(tpcb, NULL);
+					}
+					else {
+						tcp_recv(tpcb, NULL);
+						tcp_sent(tpcb, NULL);
+						tcp_poll(tpcb, NULL, 0);
+						tcp_err(tpcb,  NULL);
+					}
+					if ((err = tcp_close(tpcb)) < 0) {
+						DEBUG_ERROR("error while calling tcp_close, fd=%d, vs=%p, pcb=%p", vs->app_fd, vs, vs->pcb);
+						errno = lwip_err_to_errno(err);
+						err = -1;
+					}
+					else {
+						vs->set_state(VS_STOPPED); // success
+					}
+				}
+			}
+		}
+
+
+		// Handle transmission and reception of data
 		if (vs->socket_type == SOCK_STREAM) {
 			DEBUG_INFO("fd=%d, vs=%p, PCB=%p, vs->pcb=%p, vs->pcb->state=%d", vs->app_fd, vs, PCB, (struct tcp_pcb*)(vs->pcb), ((struct tcp_pcb*)(vs->pcb))->state);
 			if (((struct tcp_pcb*)(vs->pcb))->state == CLOSE_WAIT) {

src/lwIP.hpp

@@ -191,7 +191,7 @@ extern "C" err_t tcp_shutdown(LWIP_TCP_SHUTDOWN_SIG);
 namespace ZeroTier {
 
 	class VirtualTap;
-	struct VirtualSocket;
+	class VirtualSocket;
 
 	class lwIP
 	{

src/picoTCP.cpp

@@ -470,7 +470,7 @@ namespace ZeroTier {
 		// may now be issued in order to accept the incoming VirtualSocket from a remote host.
 		if (ev & PICO_SOCK_EV_CONN) {
 			DEBUG_EXTRA("PICO_SOCK_EV_CONN");
-			if (vs->state == ZT_SOCK_STATE_LISTENING) {
+			if (vs->state == VS_STATE_LISTENING) {
 				uint16_t port;
 				struct pico_socket *client_psock = nullptr;
 				struct pico_ip4 orig4;
@@ -529,9 +529,9 @@ namespace ZeroTier {
 				vs->_AcceptedConnections.push(new_vs);
 				new_vs->sock = new_vs->tap->_phy.wrapSocket(new_vs->sdk_fd, new_vs);
 			}
-			if (vs->state != ZT_SOCK_STATE_LISTENING) {
+			if (vs->state != VS_STATE_LISTENING) {
 				// set state so socket multiplexer logic will pick this up
-				vs->state = ZT_SOCK_STATE_UNHANDLED_CONNECTED;
+				vs->state = VS_STATE_UNHANDLED_CONNECTED;
 			}
 		}
 		// PICO_SOCK_EV_RD - triggered when new data arrives on the socket. A new receive action
@@ -882,7 +882,7 @@ namespace ZeroTier {
 				vs->picosock, err, pico_err, beautify_pico_error(pico_err));
 			return map_pico_err_to_errno(pico_err);
 		}
-		vs->state = ZT_SOCK_STATE_LISTENING;
+		vs->state = VS_STATE_LISTENING;
 		return ZT_ERR_OK;
 	}
 

src/picoTCP.hpp

@@ -86,7 +86,7 @@ namespace ZeroTier
 	int pico_eth_poll(struct pico_device *dev, int loop_score);
 
 	class VirtualTap;
-	struct VirtualSocket;
+	class VirtualSocket;
 
 	class picoTCP
 	{