Add buffer_size UDP file options

This allows UDP receive buffer size to be increased to avoid potential
packet loss result from receive buffer overrun.

Another related flag is --io_cache_size. buffer_size in UDP options
defines the UDP buffer size of the underlying system while
io_cache_size defines the size of the internal circular buffer managed
by Shaka Packager.

Closes #411

Change-Id: I57c843a88f13da546417dadc2a78df2bee0a00f3

docs/source/options/udp_file_options.rst

@@ -7,15 +7,23 @@ UDP file is of the form::
 
 Here is the list of supported options:
 
-:reuse=0|1:
+:buffer_size=<size_in_bytes>:
 
-    Allow or disallow reusing UDP sockets.
+    UDP maximum receive buffer size in bytes. Note that although it can be set
+    to any value, the actual value is capped by maximum allowed size defined by
+    the underlying operating system. On linux, the maximum size allowed can be
+    retrieved using `sysctl net.core.rmem_max` and configured using
+    `sysctl -w net.core.rmem_max=<size_in_bytes>`.
 
 :interface=<addr>:
 
     Multicast group interface address. Only the packets sent to this address are
     received. Default to "0.0.0.0" if not specified.
 
+:reuse=0|1:
+
+    Allow or disallow reusing UDP sockets.
+
 :source=<addr>:
 
     Multicast source ip address. Only the packets sent from this source address
@@ -28,3 +36,24 @@ Here is the list of supported options:
 Example::
 
     udp://224.1.2.30:88?interface=10.11.12.13&reuse=1
+
+.. note::
+
+    UDP is by definition unreliable. There could be packets dropped.
+
+    UDP packets do not get lost magically. There are things you can do to
+    minimize the packet loss. A common cause of packet loss is buffer overrun,
+    either in send buffer or receive buffer.
+
+    On Linux, you can check UDP errors by monitoring the output from
+    `netstat -suna` command.
+
+    If there is an increase in `send buffer errors` from the `netstat` output,
+    then try increasing `buffer_size` in
+    [FFmpeg](https://ffmpeg.org/ffmpeg-protocols.html#udp).
+
+    If there is an increase in `receive buffer errors`, then try increasing
+    `buffer_size` in UDP options (See above) or increasing `--io_cache_size`.
+    `buffer_size` in UDP options defines the UDP buffer size of the underlying
+    system while `io_cache_size` defines the size of the internal circular
+    buffer managed by `Shaka Packager`.

docs/source/tutorials/ffmpeg_piping.rst

@@ -13,7 +13,7 @@ There are two options to pipe data to packager.
 - UDP socket
 
     *FFmpeg* supports writing to a UDP socket and *packager* supports reading
-    from UDP sockets::
+    from UDP sockets (See :doc:`/options/udp_file_options`)::
 
         $ packager 'input=udp://127.0.0.1:40000,...' ...
         $ ffmpeg ... -f mpegts udp://127.0.0.1:40000

packager/file/udp_file.cc

@@ -273,12 +273,23 @@ bool UdpFile::Open() {
     tv.tv_sec = options->timeout_us() / 1000000;
     tv.tv_usec = options->timeout_us() % 1000000;
     if (setsockopt(new_socket.get(), SOL_SOCKET, SO_RCVTIMEO,
-                   reinterpret_cast<char*>(&tv), sizeof(tv)) < 0) {
+                   reinterpret_cast<const char*>(&tv), sizeof(tv)) < 0) {
       LOG(ERROR) << "Failed to set socket timeout.";
       return false;
     }
   }
 
+  if (options->buffer_size() > 0) {
+    const int receive_buffer_size = options->buffer_size();
+    if (setsockopt(new_socket.get(), SOL_SOCKET, SO_RCVBUF,
+                   reinterpret_cast<const char*>(&receive_buffer_size),
+                   sizeof(receive_buffer_size)) < 0) {
+      LOG(ERROR) << "Failed to set the maximum receive buffer size: "
+                 << strerror(errno);
+      return false;
+    }
+  }
+
   socket_ = new_socket.release();
   return true;
 }

packager/file/udp_options.cc

@@ -22,9 +22,10 @@ namespace {
 
 enum FieldType {
   kUnknownField = 0,
-  kReuseField,
+  kBufferSizeField,
   kInterfaceAddressField,
   kMulticastSourceField,
+  kReuseField,
   kTimeoutField,
 };
 
@@ -34,8 +35,9 @@ struct FieldNameToTypeMapping {
 };
 
 const FieldNameToTypeMapping kFieldNameTypeMappings[] = {
-    {"reuse", kReuseField},
+    {"buffer_size", kBufferSizeField},
     {"interface", kInterfaceAddressField},
+    {"reuse", kReuseField},
     {"source", kMulticastSourceField},
     {"timeout", kTimeoutField},
 };
@@ -87,6 +89,20 @@ std::unique_ptr<UdpOptions> UdpOptions::ParseFromString(
     }
     for (const auto& pair : pairs) {
       switch (GetFieldType(pair.first)) {
+        case kBufferSizeField:
+          if (!base::StringToInt(pair.second, &options->buffer_size_)) {
+            LOG(ERROR) << "Invalid udp option for buffer_size field "
+                       << pair.second;
+            return nullptr;
+          }
+          break;
+        case kInterfaceAddressField:
+          options->interface_address_ = pair.second;
+          break;
+        case kMulticastSourceField:
+          options->source_address_ = pair.second;
+          options->is_source_specific_multicast_ = true;
+          break;
         case kReuseField: {
           int reuse_value = 0;
           if (!base::StringToInt(pair.second, &reuse_value)) {
@@ -96,9 +112,6 @@ std::unique_ptr<UdpOptions> UdpOptions::ParseFromString(
           options->reuse_ = reuse_value > 0;
           break;
         }
-        case kInterfaceAddressField:
-          options->interface_address_ = pair.second;
-          break;
         case kTimeoutField:
           if (!base::StringToUint(pair.second, &options->timeout_us_)) {
             LOG(ERROR) << "Invalid udp option for timeout field "
@@ -106,10 +119,6 @@ std::unique_ptr<UdpOptions> UdpOptions::ParseFromString(
             return nullptr;
           }
           break;
-        case kMulticastSourceField:
-          options->source_address_ = pair.second;
-          options->is_source_specific_multicast_ = true;
-          break;
         default:
           LOG(ERROR) << "Unknown field in udp options (\"" << pair.first
                      << "\").";

packager/file/udp_options.h

@@ -30,22 +30,28 @@ class UdpOptions {
   bool is_source_specific_multicast() const {
     return is_source_specific_multicast_;
   }
+  int buffer_size() const { return buffer_size_; }
 
  private:
   UdpOptions() = default;
 
-  /// IP Address.
+  // IP Address.
   std::string address_ = "0.0.0.0";
   uint16_t port_ = 0;
-  /// Allow or disallow reusing UDP sockets.
+  // Allow or disallow reusing UDP sockets.
   bool reuse_ = false;
   // Address of the interface over which to receive UDP multicast streams.
   std::string interface_address_ = "0.0.0.0";
-  /// Timeout in microseconds. 0 to indicate unlimited timeout.
+  // Timeout in microseconds. 0 to indicate unlimited timeout.
   unsigned timeout_us_ = 0;
   // Source specific multicast source address
   std::string source_address_ = "0.0.0.0";
   bool is_source_specific_multicast_ = false;
+  // Maximum receive buffer size in bytes.
+  // Note that the actual buffer size is capped by the maximum buffer size set
+  // by the underlying operating system ('sysctl net.core.rmem_max' on Linux
+  // returns the maximum receive memory size).
+  int buffer_size_ = 0;
 };
 
 }  // namespace shaka

packager/file/udp_options_unittest.cc

@@ -117,4 +117,9 @@ TEST_F(UdpOptionsTest, InvalidTimeout) {
       "224.1.2.30:88?interface=10.11.12.13&timeout=1a9"));
 }
 
+TEST_F(UdpOptionsTest, BufferSize) {
+  auto options = UdpOptions::ParseFromString("224.1.2.30:88?buffer_size=1234");
+  EXPECT_EQ(1234, options->buffer_size());
+}
+
 }  // namespace shaka