Fix slow websockets performance when sending large messages.

Closes #1390. Thanks to aalibasic.

ChangeLog.txt

@@ -2,6 +2,7 @@ Broker:
 - Fix v5 DISCONNECT packets with remaining length == 2 being treated as a
   protocol error. Closes #1367.
 - Fix support for libwebsockets 3.x.
+- Fix slow websockets performance when sending large messages. Closes #1390.
 
 Build:
 - Fix missing function warnings on NetBSD.

src/websockets.c

@@ -47,6 +47,12 @@ POSSIBILITY OF SUCH DAMAGE.
 #  include <sys/socket.h>
 #endif
 
+/* Be careful if changing these, if TX is not bigger than SERV then there can
+ * be very large write performance penalties.
+ */
+#define WS_SERV_BUF_SIZE 4096
+#define WS_TX_BUF_SIZE (WS_SERV_BUF_SIZE*2)
+
 extern struct mosquitto_db int_db;
 
 #if defined(LWS_LIBRARY_VERSION_NUMBER)
@@ -97,7 +103,7 @@ static struct libwebsocket_protocols protocols[] = {
 #ifdef LWS_LIBRARY_VERSION_NUMBER
 		NULL,								/* user v1.4 on */
 #  if LWS_LIBRARY_VERSION_NUMBER >= 2003000
-		0									/* tx_packet_size v2.3.0 */
+		WS_TX_BUF_SIZE						/* tx_packet_size v2.3.0 */
 #  endif
 #endif
 	},
@@ -115,7 +121,7 @@ static struct libwebsocket_protocols protocols[] = {
 #ifdef LWS_LIBRARY_VERSION_NUMBER
 		NULL,								/* user v1.4 on */
 #  if LWS_LIBRARY_VERSION_NUMBER >= 2003000
-		0									/* tx_packet_size v2.3.0 */
+		WS_TX_BUF_SIZE						/* tx_packet_size v2.3.0 */
 #  endif
 #endif
 	},
@@ -133,7 +139,7 @@ static struct libwebsocket_protocols protocols[] = {
 #ifdef LWS_LIBRARY_VERSION_NUMBER
 		NULL,								/* user v1.4 on */
 #  if LWS_LIBRARY_VERSION_NUMBER >= 2003000
-		0									/* tx_packet_size v2.3.0 */
+		WS_TX_BUF_SIZE						/* tx_packet_size v2.3.0 */
 #  endif
 #endif
 	},
@@ -180,6 +186,7 @@ static int callback_mqtt(struct libwebsocket_context *context,
 	struct mosquitto_db *db;
 	struct mosquitto *mosq = NULL;
 	struct mosquitto__packet *packet;
+	size_t txlen;
 	int count;
 	const struct libwebsocket_protocols *p;
 	struct libws_mqtt_data *u = (struct libws_mqtt_data *)user;
@@ -287,7 +294,12 @@ static int callback_mqtt(struct libwebsocket_context *context,
 					memmove(&packet->payload[LWS_SEND_BUFFER_PRE_PADDING], packet->payload, packet->packet_length);
 					packet->pos += LWS_SEND_BUFFER_PRE_PADDING;
 				}
-				count = libwebsocket_write(wsi, &packet->payload[packet->pos], packet->to_process, LWS_WRITE_BINARY);
+				if(packet->to_process > WS_TX_BUF_SIZE){
+					txlen = WS_TX_BUF_SIZE;
+				}else{
+					txlen = packet->to_process;
+				}
+				count = libwebsocket_write(wsi, &packet->payload[packet->pos], txlen, LWS_WRITE_BINARY);
 				if(count < 0){
 					if (mosq->state == mosq_cs_disconnect_ws || mosq->state == mosq_cs_disconnecting){
 						return -1;
@@ -757,6 +769,7 @@ struct libwebsocket_context *mosq_websockets_init(struct mosquitto__listener *li
 	}
 
 	info.user = user;
+	info.pt_serv_buf_size = WS_SERV_BUF_SIZE;
 	listener->ws_protocol = p;
 
 	lws_set_log_level(conf->websockets_log_level, log_wrap);