Socket library and Config library restructuring, improvement to UDP socket reliability

lib/certificate.cpp

@@ -1,5 +1,7 @@
 #include "certificate.h"
 #include "defines.h"
+#include <string.h>
+
 
 Certificate::Certificate() : rsa_ctx(NULL){
   memset((void *)&cert, 0x00, sizeof(cert));

lib/comms.cpp

@@ -5,6 +5,8 @@
 #include "encode.h"
 #include "procs.h"
 #include "timing.h"
+#include <string.h>
+#include <fcntl.h>
 
 namespace Comms{
   Comms::Comms(){

lib/config.cpp

@@ -38,8 +38,8 @@
 bool Util::Config::is_active = false;
 bool Util::Config::is_restarting = false;
 static Socket::Server *serv_sock_pointer = 0;
-uint32_t Util::Config::printDebugLevel = DEBUG; //
-std::string Util::Config::streamName;
+uint32_t Util::printDebugLevel = DEBUG;
+std::string Util::streamName;
 char Util::exitReason[256] = {0};
 
 void Util::logExitReason(const char *format, ...){

lib/config.h

@@ -13,6 +13,8 @@
 
 /// Contains utility code, not directly related to streaming media
 namespace Util{
+  extern uint32_t printDebugLevel;
+  extern std::string streamName; ///< Used by debug messages to identify the stream name
   extern char exitReason[256];
   void logExitReason(const char * format, ...);
 
@@ -27,8 +29,6 @@ namespace Util{
     // variables
     static bool is_active;     ///< Set to true by activate(), set to false by the signal handler.
     static bool is_restarting; ///< Set to true when restarting, set to false on boot.
-    static uint32_t printDebugLevel;
-    static std::string streamName; ///< Used by debug messages to identify the stream name
     // functions
     Config();
     Config(std::string cmd);

lib/defines.h

@@ -23,11 +23,18 @@
 
 #define APPIDENT APPNAME "/" PACKAGE_VERSION
 #define __STDC_FORMAT_MACROS 1
-#include "config.h"
 #include <inttypes.h>
 #include <stdint.h>
 #include <stdio.h>
 #include <unistd.h>
+#include <string>
+
+//Declare as extern so we don't have to include the whole config.h header
+namespace Util{
+  extern uint32_t printDebugLevel;
+  extern std::string streamName;
+}
+
 static const char *DBG_LVL_LIST[] ={"NONE", "FAIL",     "ERROR",   "WARN",   "INFO",    "MEDIUM",
                                      "HIGH", "VERYHIGH", "EXTREME", "INSANE", "DONTEVEN"};
 
@@ -44,29 +51,29 @@ static const char *DBG_LVL_LIST[] ={"NONE", "FAIL",     "ERROR",   "WARN",   "IN
 
 #if DEBUG >= DLVL_DEVEL
 #define DEBUG_MSG(lvl, msg, ...)                                                                     \
-  if (Util::Config::printDebugLevel >= lvl){\
+  if (Util::printDebugLevel >= lvl){\
     fprintf(stderr, "%s|%s|%d|%s:%d|%s|" msg "\n", DBG_LVL_LIST[lvl], program_invocation_short_name, \
-            getpid(), __FILE__, __LINE__, Util::Config::streamName.c_str(), ##__VA_ARGS__);          \
+            getpid(), __FILE__, __LINE__, Util::streamName.c_str(), ##__VA_ARGS__);          \
   }
 #else
 #define DEBUG_MSG(lvl, msg, ...)                                                                   \
-  if (Util::Config::printDebugLevel >= lvl){\
+  if (Util::printDebugLevel >= lvl){\
     fprintf(stderr, "%s|%s|%d||%s|" msg "\n", DBG_LVL_LIST[lvl], program_invocation_short_name,    \
-            getpid(), Util::Config::streamName.c_str(), ##__VA_ARGS__);                            \
+            getpid(), Util::streamName.c_str(), ##__VA_ARGS__);                            \
   }
 #endif
 #else
 #if DEBUG >= DLVL_DEVEL
 #define DEBUG_MSG(lvl, msg, ...)                                                                   \
-  if (Util::Config::printDebugLevel >= lvl){\
+  if (Util::printDebugLevel >= lvl){\
     fprintf(stderr, "%s|MistProcess|%d|%s:%d|%s|" msg "\n", DBG_LVL_LIST[lvl], getpid(), __FILE__, \
-            __LINE__, Util::Config::streamName.c_str(), ##__VA_ARGS__);                            \
+            __LINE__, Util::streamName.c_str(), ##__VA_ARGS__);                            \
   }
 #else
 #define DEBUG_MSG(lvl, msg, ...)                                                                   \
-  if (Util::Config::printDebugLevel >= lvl){\
+  if (Util::printDebugLevel >= lvl){\
     fprintf(stderr, "%s|MistProcess|%d||%s|" msg "\n", DBG_LVL_LIST[lvl], getpid(),                \
-            Util::Config::streamName.c_str(), ##__VA_ARGS__);                                      \
+            Util::streamName.c_str(), ##__VA_ARGS__);                                      \
   }
 #endif
 #endif

lib/dtls_srtp_handshake.cpp

@@ -1,6 +1,7 @@
 #include "defines.h"
 #include "dtls_srtp_handshake.h"
 #include <algorithm>
+#include <string.h>
 
 /* Write mbedtls into a log file. */
 #define LOG_TO_FILE 0

lib/http_parser.cpp

@@ -8,6 +8,7 @@
 #include "timing.h"
 #include "url.h"
 #include "util.h"
+#include "json.h"
 #include <iomanip>
 #include <strings.h>
 

lib/procs.cpp

@@ -13,6 +13,7 @@
 #include <wait.h>
 #endif
 #include "timing.h"
+#include "json.h"
 #include <errno.h>
 #include <fcntl.h>
 #include <iostream>

lib/riff.cpp

@@ -1,4 +1,5 @@
 #include "riff.h"
+#include <string.h>
 
 namespace RIFF{
 

lib/riff.h

@@ -2,6 +2,7 @@
 #include "bitfields.h"
 #include "defines.h"
 #include <string>
+#include <ostream>
 
 namespace RIFF{
 

lib/shared_memory.cpp

@@ -13,7 +13,6 @@
 #include <iostream>
 #include <sys/mman.h>
 #include <sys/sem.h>
-#include <sys/stat.h>
 #include <unistd.h>
 
 #if defined(__CYGWIN__) || defined(_WIN32)

lib/shared_memory.h

@@ -1,6 +1,7 @@
 #pragma once
 #include <set>
 #include <string>
+#include <sys/stat.h>
 
 #include "defines.h"
 #include "timing.h"

lib/socket.cpp

@@ -5,6 +5,7 @@
 #include "defines.h"
 #include "socket.h"
 #include "timing.h"
+#include "json.h"
 #include <cstdlib>
 #include <ifaddrs.h>
 #include <netdb.h>
@@ -114,7 +115,7 @@ bool Socket::isLocal(const std::string &remotehost){
 /// Helper function that matches two binary-format IPv6 addresses with prefix bits of prefix.
 bool Socket::matchIPv6Addr(const std::string &A, const std::string &B, uint8_t prefix){
   if (!prefix){prefix = 128;}
-  if (Util::Config::printDebugLevel >= DLVL_MEDIUM){
+  if (Util::printDebugLevel >= DLVL_MEDIUM){
     std::string Astr, Bstr;
     Socket::hostBytesToStr(A.data(), 16, Astr);
     Socket::hostBytesToStr(B.data(), 16, Bstr);
@@ -1572,17 +1573,56 @@ Socket::UDPConnection::UDPConnection(bool nonblock){
     sock = socket(AF_INET, SOCK_DGRAM, 0);
     family = AF_INET;
   }
-  if (sock == -1){FAIL_MSG("Could not create UDP socket: %s", strerror(errno));}
+  if (sock == -1){
+    FAIL_MSG("Could not create UDP socket: %s", strerror(errno));
+  }else{
+    if (nonblock){setBlocking(!nonblock);}
+    checkRecvBuf();
+  }
   up = 0;
   down = 0;
   destAddr = 0;
   destAddr_size = 0;
-  data = 0;
-  data_size = 0;
-  data_len = 0;
-  if (nonblock){setBlocking(!nonblock);}
+  data.allocate(2048);
 }// Socket::UDPConnection UDP Contructor
 
+///Checks if the UDP receive buffer is at least 1 mbyte, attempts to increase and warns user through log message on failure.
+void Socket::UDPConnection::checkRecvBuf(){
+  if (sock == -1){return;}
+  int recvbuf = 0;
+  int origbuf = 0;
+  socklen_t slen = sizeof(recvbuf);
+  getsockopt(sock, SOL_SOCKET, SO_RCVBUF, (void*)&recvbuf, &slen);
+  origbuf = recvbuf;
+  if (recvbuf < 1024*1024){
+    recvbuf = 1024*1024;
+    setsockopt(sock, SOL_SOCKET, SO_RCVBUF, (void*)&recvbuf, sizeof(recvbuf));
+    slen = sizeof(recvbuf);
+    getsockopt(sock, SOL_SOCKET, SO_RCVBUF, (void*)&recvbuf, &slen);
+    if (recvbuf < 1024*1024){
+      recvbuf = 1024*1024;
+      setsockopt(sock, SOL_SOCKET, SO_RCVBUFFORCE, (void*)&recvbuf, sizeof(recvbuf));
+      slen = sizeof(recvbuf);
+      getsockopt(sock, SOL_SOCKET, SO_RCVBUF, (void*)&recvbuf, &slen);
+    }
+    if (recvbuf < 200*1024){
+      recvbuf = 200*1024;
+      setsockopt(sock, SOL_SOCKET, SO_RCVBUF, (void*)&recvbuf, sizeof(recvbuf));
+      slen = sizeof(recvbuf);
+      getsockopt(sock, SOL_SOCKET, SO_RCVBUF, (void*)&recvbuf, &slen);
+      if (recvbuf < 200*1024){
+        recvbuf = 200*1024;
+        setsockopt(sock, SOL_SOCKET, SO_RCVBUFFORCE, (void*)&recvbuf, sizeof(recvbuf));
+        slen = sizeof(recvbuf);
+        getsockopt(sock, SOL_SOCKET, SO_RCVBUF, (void*)&recvbuf, &slen);
+      }
+    }
+    if (recvbuf < 200*1024){
+      WARN_MSG("Your UDP receive buffer is set < 200 kbyte (%db) and the kernel denied our request for an increase. It's recommended to set your net.core.rmem_max setting to at least 200 kbyte for best results.", origbuf);
+    }
+  }
+}
+
 /// Copies a UDP socket, re-allocating local copies of any needed structures.
 /// The data/data_size/data_len variables are *not* copied over.
 Socket::UDPConnection::UDPConnection(const UDPConnection &o){
@@ -1594,6 +1634,7 @@ Socket::UDPConnection::UDPConnection(const UDPConnection &o){
     family = AF_INET;
   }
   if (sock == -1){FAIL_MSG("Could not create UDP socket: %s", strerror(errno));}
+  checkRecvBuf();
   up = 0;
   down = 0;
   if (o.destAddr && o.destAddr_size){
@@ -1604,13 +1645,7 @@ Socket::UDPConnection::UDPConnection(const UDPConnection &o){
     destAddr = 0;
     destAddr_size = 0;
   }
-  data = (char *)malloc(1024);
-  if (data){
-    data_size = 1024;
-  }else{
-    data_size = 0;
-  }
-  data_len = 0;
+  data.allocate(2048);
 }
 
 /// Close the UDP socket
@@ -1629,10 +1664,6 @@ Socket::UDPConnection::~UDPConnection(){
     free(destAddr);
     destAddr = 0;
   }
-  if (data){
-    free(data);
-    data = 0;
-  }
 }
 
 /// Stores the properties of the receiving end of this UDP socket.
@@ -1674,6 +1705,7 @@ void Socket::UDPConnection::SetDestination(std::string destIp, uint32_t port){
       close();
       family = rp->ai_family;
       sock = socket(family, SOCK_DGRAM, 0);
+      checkRecvBuf();
       if (boundPort){
         INFO_MSG("Rebinding to %s:%d %s", boundAddr.c_str(), boundPort, boundMulti.c_str());
         bind(boundPort, boundAddr, boundMulti);
@@ -1815,6 +1847,7 @@ uint16_t Socket::UDPConnection::bind(int port, std::string iface, const std::str
   for (rp = addr_result; rp != NULL; rp = rp->ai_next){
     sock = socket(rp->ai_family, rp->ai_socktype, rp->ai_protocol);
     if (sock == -1){continue;}
+    checkRecvBuf();
     char human_addr[INET6_ADDRSTRLEN];
     char human_port[16];
     getnameinfo(rp->ai_addr, rp->ai_addrlen, human_addr, INET6_ADDRSTRLEN, human_port, 16,
@@ -1964,36 +1997,23 @@ uint16_t Socket::UDPConnection::bind(int port, std::string iface, const std::str
 bool Socket::UDPConnection::Receive(){
   if (sock == -1){return false;}
 #ifdef __CYGWIN__
-  if (data_size != SOCKETSIZE){
-    data = (char *)realloc(data, SOCKETSIZE);
-    data_size = SOCKETSIZE;
-  }
+    data.allocate((SOCKETSIZE);
 #endif
-  int r = recvfrom(sock, data, data_size, MSG_PEEK | MSG_TRUNC | MSG_DONTWAIT, 0, 0);
+  data.truncate(0);
+  socklen_t destsize = destAddr_size;
+  int r = recvfrom(sock, data, data.rsize(), MSG_TRUNC | MSG_DONTWAIT, (sockaddr *)destAddr, &destsize);
   if (r == -1){
     if (errno != EAGAIN){INFO_MSG("UDP receive: %d (%s)", errno, strerror(errno));}
-    data_len = 0;
     return false;
   }
-  if (data_size < (unsigned int)r){
-    char *tmp = (char *)realloc(data, r);
-    if (tmp){
-      data = tmp;
-      data_size = r;
-    }else{
-      FAIL_MSG("Could not resize socket buffer to %d bytes!", r);
-      return false;
-    }
+  data.append(0, r);
+  down += r;
+  //Handle UDP packets that are too large
+  if (data.rsize() < (unsigned int)r){
+    INFO_MSG("Doubling UDP socket buffer from %" PRIu32 " to %" PRIu32, data.rsize(), data.rsize()*2);
+    data.allocate(data.rsize()*2);
   }
-  socklen_t destsize = destAddr_size;
-  r = recvfrom(sock, data, data_size, 0, (sockaddr *)destAddr, &destsize);
-  if (r > 0){
-    down += r;
-    data_len = r;
-    return true;
-  }
-  data_len = 0;
-  return false;
+  return (r > 0);
 }
 
 int Socket::UDPConnection::getSock(){

lib/socket.h

@@ -15,6 +15,7 @@
 #include <sys/types.h>
 #include <sys/un.h>
 #include <unistd.h>
+#include "util.h"
 
 #ifdef SSL
 #include "mbedtls/ctr_drbg.h"
@@ -195,14 +196,13 @@ namespace Socket{
     unsigned int destAddr_size; ///< Size of the destination address pointer.
     unsigned int up;            ///< Amount of bytes transferred up.
     unsigned int down;          ///< Amount of bytes transferred down.
-    unsigned int data_size;     ///< The size in bytes of the allocated space in the data pointer.
     int family;                 ///< Current socket address family
     std::string boundAddr, boundMulti;
     int boundPort;
+    void checkRecvBuf();
 
   public:
-    char *data;            ///< Holds the last received packet.
-    unsigned int data_len; ///< The size in bytes of the last received packet.
+    Util::ResizeablePointer data;
     UDPConnection(const UDPConnection &o);
     UDPConnection(bool nonblock = false);
     ~UDPConnection();

lib/srtp.cpp

@@ -1,6 +1,7 @@
 #include "defines.h"
 #include "srtp.h"
 #include <algorithm>
+#include <string.h>
 
 /* --------------------------------------- */
 

lib/srtp.h

@@ -3,6 +3,7 @@
 #include <srtp2/srtp.h>
 #include <stdint.h>
 #include <string>
+#include <vector>
 
 #define SRTP_PARSER_MASTER_KEY_LEN 16
 #define SRTP_PARSER_MASTER_SALT_LEN 14

lib/stream.cpp

@@ -407,8 +407,8 @@ bool Util::startInput(std::string streamname, std::string filename, bool forkFir
                     (char *)filename.c_str()};
   int argNum = 3;
   std::string debugLvl;
-  if (Util::Config::printDebugLevel != DEBUG && !str_args.count("--debug")){
-    debugLvl = JSON::Value(Util::Config::printDebugLevel).asString();
+  if (Util::printDebugLevel != DEBUG && !str_args.count("--debug")){
+    debugLvl = JSON::Value(Util::printDebugLevel).asString();
     argv[++argNum] = (char *)"--debug";
     argv[++argNum] = (char *)debugLvl.c_str();
   }
@@ -563,7 +563,7 @@ pid_t Util::startPush(const std::string &streamname, std::string &target, int de
   // Set original target string in environment
   setenv("MST_ORIG_TARGET", target.c_str(), 1);
   //If no debug level set, default to level of starting process
-  if (debugLvl < 0){debugLvl = Util::Config::printDebugLevel;}
+  if (debugLvl < 0){debugLvl = Util::printDebugLevel;}
 
   // The target can hold variables like current time etc
   streamVariables(target, streamname);
@@ -1285,7 +1285,7 @@ std::set<size_t> Util::wouldSelect(const DTSC::Meta &M, const std::map<std::stri
     }
   }
 
-  if (Util::Config::printDebugLevel >= DLVL_MEDIUM){
+  if (Util::printDebugLevel >= DLVL_MEDIUM){
     // print the selected tracks
     std::stringstream selected;
     for (std::set<size_t>::iterator it = result.begin(); it != result.end(); it++){

lib/stun.cpp

@@ -1,6 +1,7 @@
 #include "checksum.h" // for crc32
 #include "defines.h"
 #include "stun.h"
+#include "socket.h"
 
 /* --------------------------------------- */
 

lib/triggers.cpp

@@ -24,6 +24,7 @@
 #include "timing.h"
 #include "triggers.h"
 #include "util.h"
+#include "json.h"
 #include <string.h> //for strncmp
 
 namespace Triggers{