From: Jaroslav Kysela <perex@perex.cz>
Date: Tue, 30 Oct 2018 18:55:59 +0000 (+0100)
Subject: htsp server: improve the htsp streaming connection limit check, issue #5290
X-Git-Url: http://git.ipfire.org/cgi-bin/gitweb.cgi?a=commitdiff_plain;h=ed33294f9cdfe41696e9e95cf81a75510d6f1193;p=thirdparty%2Ftvheadend.git

htsp server: improve the htsp streaming connection limit check, issue #5290
---

diff --git a/src/htsp_server.c b/src/htsp_server.c
index 6d1a7e07d..bc1ce16ce 100644
--- a/src/htsp_server.c
+++ b/src/htsp_server.c
@@ -3223,7 +3223,7 @@ static int
 htsp_read_loop(htsp_connection_t *htsp)
 {
   htsmsg_t *m = NULL, *reply;
-  int r = 0, i;
+  int run = 1, r = 0, i, streaming = 0;
   const char *method;
   void *tcp_id = NULL;;
 
@@ -3238,7 +3238,7 @@ htsp_read_loop(htsp_connection_t *htsp)
   htsp->htsp_granted_access = access_get_by_addr(htsp->htsp_peer);
   htsp->htsp_granted_access->aa_rights |= ACCESS_HTSP_INTERFACE;
 
-  tcp_id = tcp_connection_launch(htsp->htsp_fd, htsp_server_status,
+  tcp_id = tcp_connection_launch(htsp->htsp_fd, streaming, htsp_server_status,
                                  htsp->htsp_granted_access);
 
   pthread_mutex_unlock(&global_lock);
@@ -3250,7 +3250,7 @@ htsp_read_loop(htsp_connection_t *htsp)
 
   /* Session main loop */
 
-  while(tvheadend_is_running()) {
+  while(run && tvheadend_is_running()) {
 readmsg:
     reply = NULL;
 
@@ -3260,12 +3260,14 @@ readmsg:
     pthread_mutex_lock(&global_lock);
     if (htsp_authenticate(htsp, m)) {
       tcp_connection_land(tcp_id);
-      tcp_id = tcp_connection_launch(htsp->htsp_fd, htsp_server_status,
+      tcp_id = tcp_connection_launch(htsp->htsp_fd, streaming, htsp_server_status,
                                      htsp->htsp_granted_access);
       if (tcp_id == NULL) {
-        htsmsg_destroy(m);
-        pthread_mutex_unlock(&global_lock);
-        return 1;
+        reply = htsmsg_create_map();
+        htsmsg_add_u32(reply, "noaccess", 1);
+        htsmsg_add_u32(reply, "connlimit", 1);
+        run = 0;
+        goto send_reply_with_unlock;
       }
     }
 
@@ -3292,6 +3294,18 @@ readmsg:
             goto readmsg;
 
           } else {
+            if (!strcmp(method, "subscribe") && !streaming) {
+              tcp_connection_land(tcp_id);
+              tcp_id = tcp_connection_launch(htsp->htsp_fd, 1, htsp_server_status,
+                                             htsp->htsp_granted_access);
+              if (tcp_id == NULL) {
+                reply = htsmsg_create_map();
+                htsmsg_add_u32(reply, "noaccess", 1);
+                htsmsg_add_u32(reply, "connlimit", 1);
+                goto send_reply_with_unlock;
+              }
+              streaming = 1;
+            }
             reply = htsp_methods[i].fn(htsp, m);
           }
           break;
@@ -3306,6 +3320,7 @@ readmsg:
       reply = htsp_error(htsp, N_("Invalid arguments"));
     }
 
+send_reply_with_unlock:
     pthread_mutex_unlock(&global_lock);
 
     if(reply != NULL) /* Methods can do all the replying inline */
diff --git a/src/satip/rtsp.c b/src/satip/rtsp.c
index d0af82c8b..6aa2c7527 100644
--- a/src/satip/rtsp.c
+++ b/src/satip/rtsp.c
@@ -1724,7 +1724,7 @@ rtsp_serve(int fd, void **opaque, struct sockaddr_storage *peer,
   tcp_get_str_from_ip(peer, buf + strlen(buf), sizeof(buf) - strlen(buf));
   aa.aa_representative = buf;
 
-  tcp = tcp_connection_launch(fd, rtsp_stream_status, &aa);
+  tcp = tcp_connection_launch(fd, 1, rtsp_stream_status, &aa);
 
   /* Note: global_lock held on entry */
   pthread_mutex_unlock(&global_lock);
diff --git a/src/tcp.c b/src/tcp.c
index e38f189be..62256c613 100644
--- a/src/tcp.c
+++ b/src/tcp.c
@@ -530,6 +530,7 @@ typedef struct tcp_server_launch {
   pthread_t tid;
   uint32_t id;
   int fd;
+  int streaming;
   tcp_server_ops_t ops;
   void *opaque;
   char *representative;
@@ -572,10 +573,10 @@ tcp_connection_count(access_t *aa)
  */
 void *
 tcp_connection_launch
-  (int fd, void (*status) (void *opaque, htsmsg_t *m), access_t *aa)
+  (int fd, int streaming, void (*status) (void *opaque, htsmsg_t *m), access_t *aa)
 {
   tcp_server_launch_t *tsl, *res;
-  uint32_t used, used2;
+  uint32_t sused, used2;
   int64_t started = mclk();
   int c1, c2;
 
@@ -588,7 +589,7 @@ tcp_connection_launch
 
 try_again:
   res = NULL;
-  used = 0;
+  sused = 0;
   LIST_FOREACH(tsl, &tcp_server_active, alink) {
     if (tsl->fd == fd) {
       res = tsl;
@@ -597,7 +598,8 @@ try_again:
       continue;
     }
     if (!strcmp(aa->aa_representative ?: "", tsl->representative ?: ""))
-      used++;
+      if (tsl->streaming)
+        sused++;
   }
   if (res == NULL)
     return NULL;
@@ -605,8 +607,8 @@ try_again:
   if (aa->aa_conn_limit || aa->aa_conn_limit_streaming) {
     used2 = aa->aa_conn_limit ? dvr_usage_count(aa) : 0;
     /* the rule is: allow if one condition is OK */
-    c1 = aa->aa_conn_limit ? used + used2 >= aa->aa_conn_limit : -1;
-    c2 = aa->aa_conn_limit_streaming ? used >= aa->aa_conn_limit_streaming : -1;
+    c1 = aa->aa_conn_limit ? sused + used2 >= aa->aa_conn_limit : -1;
+    c2 = aa->aa_conn_limit_streaming ? sused >= aa->aa_conn_limit_streaming : -1;
 
     if (c1 && c2) {
       if (started + sec2mono(5) < mclk()) {
@@ -614,7 +616,7 @@ try_again:
                         "(limit %u, streaming limit %u, active streaming %u, DVR %u)",
                  aa->aa_username ?: "", aa->aa_representative ?: "",
                  aa->aa_conn_limit, aa->aa_conn_limit_streaming,
-                 used, used2);
+                 sused, used2);
         return NULL;
       }
       pthread_mutex_unlock(&global_lock);
@@ -628,6 +630,7 @@ try_again:
 
   res->representative = aa->aa_representative ? strdup(aa->aa_representative) : NULL;
   res->status = status;
+  res->streaming = streaming;
   LIST_INSERT_HEAD(&tcp_server_launches, res, link);
   notify_reload("connections");
   return res;
@@ -1135,6 +1138,7 @@ tcp_server_connections ( void )
     htsmsg_add_str(e, "peer", buf);
     htsmsg_add_u32(e, "peer_port", ntohs(IP_PORT(tsl->peer)));
     htsmsg_add_s64(e, "started", tsl->started);
+    htsmsg_add_u32(e, "streaming", tsl->streaming);
     tsl->status(tsl->opaque, e);
     htsmsg_add_msg(l, NULL, e);
   }
diff --git a/src/tcp.h b/src/tcp.h
index ad6ef018f..c88c3b19b 100644
--- a/src/tcp.h
+++ b/src/tcp.h
@@ -155,7 +155,8 @@ int tcp_socket_dead(int fd);
 struct access;
 
 uint32_t tcp_connection_count(struct access *aa);
-void *tcp_connection_launch(int fd, void (*status) (void *opaque, htsmsg_t *m),
+void *tcp_connection_launch(int fd, int streaming,
+                            void (*status) (void *opaque, htsmsg_t *m),
                             struct access *aa);
 void tcp_connection_land(void *tcp_id);
 void tcp_connection_cancel(uint32_t id);
diff --git a/src/webui/static/app/status.js b/src/webui/static/app/status.js
index 6842b1d38..8e85646be 100644
--- a/src/webui/static/app/status.js
+++ b/src/webui/static/app/status.js
@@ -655,7 +655,8 @@ tvheadend.status_conns = function(panel, index) {
                     type: 'date',
                     dateFormat: 'U', /* unix time */
                     sortType: Ext.data.SortTypes.asDate
-                }
+                },
+                { name: 'streaming' }
             ],
             url: 'api/status/connections',
             autoLoad: true,
@@ -719,6 +720,12 @@ tvheadend.status_conns = function(panel, index) {
                 dataIndex: 'started',
                 renderer: renderDate,
                 sortable: true
+            }, {
+                width: 50,
+                id: 'streaming',
+                header: _("Streaming"),
+                dataIndex: 'streaming',
+                sortable: true
             }, {
                 width: 50,
                 id: 'server',
diff --git a/src/webui/webui.c b/src/webui/webui.c
index e59cb4376..feb455096 100644
--- a/src/webui/webui.c
+++ b/src/webui/webui.c
@@ -327,7 +327,7 @@ http_stream_status ( void *opaque, htsmsg_t *m )
 static inline void *
 http_stream_preop ( http_connection_t *hc )
 {
-  return tcp_connection_launch(hc->hc_fd, http_stream_status, hc->hc_access);
+  return tcp_connection_launch(hc->hc_fd, 1, http_stream_status, hc->hc_access);
 }
 
 static inline void