Support SO_MAX_PACING_RATE.

SO_MAX_PACING_RATE is the newfangled socket option from Eric Dumazet,
used with the new fq packet scheduler in Linux. It allows you to set
a max rate for the socket (presumably a stricter upper bound than the
RTT-based estimate from the kernel), delivering pacing without having
to resort to the relatively complex mark setup. It seems to enter
the Linux kernel in 3.13 at the earliest; not unlikely even later.

In time, fwmark will be deprecated, but the implementation of TCP pacing in
Linux is still a bit shaky (especially with not-always-filling applications
like streaming), so fwmark will stay the primary solution for now.

diff --git a/README b/README
index b965d03ac2142844d66aac59fd91748f28378ea7..4d5d68012106dc25f63e75039a70c6116f2f6173 100644 (file)
--- a/README
+++ b/README
@@ -13,6 +13,9 @@ A short list of features:
    (unless you delete the stream they are watching, of course).
    Cubemap also survives the encoder dying and reconnecting.
  - Per-stream fwmark support, for TCP pacing through tc (separate config needed).
    (unless you delete the stream they are watching, of course).
    Cubemap also survives the encoder dying and reconnecting.
  - Per-stream fwmark support, for TCP pacing through tc (separate config needed).

+ - Support for setting max pacing rate through the fq packet scheduler
+   (obsoletes the previous point, but depends on experimental kernel patches
+   that will hit Linux in 3.13 at the earliest)

  - Reflects anything VLC can reflect over HTTP, even the muxes VLC
    has problems reflecting itself (in particular, FLV).
  - IPv4 support. Yes, Cubemap even supports (some) legacy protocols.
  - Reflects anything VLC can reflect over HTTP, even the muxes VLC
    has problems reflecting itself (in particular, FLV).
  - IPv4 support. Yes, Cubemap even supports (some) legacy protocols.

diff --git a/client.cpp b/client.cpp
index f6361f34ab47e62dc30ec1ebd6844ae73462d21a..eea41a5e0da43d40b0c468b206158f69198ac1b6 100644 (file)
--- a/client.cpp
+++ b/client.cpp
@@ -9,6 +9,10 @@
 #include "state.pb.h"
 #include "stream.h"
 
 #include "state.pb.h"
 #include "stream.h"
 

+#ifndef SO_MAX_PACING_RATE
+#define SO_MAX_PACING_RATE 47
+#endif
+

 using namespace std;
 
 Client::Client(int sock)
 using namespace std;
 
 Client::Client(int sock)


@@ -80,6 +84,11 @@ Client::Client(const ClientProto &serialized, Stream *stream)
                }
                fwmark = 0;
        }
                }
                fwmark = 0;
        }

+       if (setsockopt(sock, SOL_SOCKET, SO_MAX_PACING_RATE, &stream->pacing_rate, sizeof(stream->pacing_rate)) == -1) {
+               if (stream->pacing_rate != ~0U) {
+                       log_perror("setsockopt(SO_MAX_PACING_RATE)");
+               }
+       }

 }
 
 ClientProto Client::serialize() const
 }
 
 ClientProto Client::serialize() const

diff --git a/config.cpp b/config.cpp
index cb16cbc33e06f7d1032d2b8d13f20379b20c262a..f5182aac1512e9b8307ac57d16ada0b639260257 100644 (file)
--- a/config.cpp
+++ b/config.cpp
@@ -234,6 +234,14 @@ bool parse_stream(const ConfigLine &line, Config *config)
                stream.mark_pool = allocate_mark_pool(from, to, config);
        }
 
                stream.mark_pool = allocate_mark_pool(from, to, config);
        }
 

+       // Parse the pacing rate, converting from kilobits to bytes as needed.
+       map<string, string>::const_iterator pacing_rate_it = line.parameters.find("pacing_rate_kbit");
+       if (pacing_rate_it == line.parameters.end()) {
+               stream.pacing_rate = ~0U;
+       } else {
+               stream.pacing_rate = atoi(pacing_rate_it->second.c_str()) * 1024 / 8;
+       }
+

        config->streams.push_back(stream);
        return true;
 }
        config->streams.push_back(stream);
        return true;
 }


@@ -316,9 +324,18 @@ bool parse_udpstream(const ConfigLine &line, Config *config)
                udpstream.mark_pool = allocate_mark_pool(from, to, config);
        }
 
                udpstream.mark_pool = allocate_mark_pool(from, to, config);
        }
 

+       // Parse the pacing rate, converting from kilobits to bytes as needed.
+       map<string, string>::const_iterator pacing_rate_it = line.parameters.find("pacing_rate_kbit");
+       if (pacing_rate_it == line.parameters.end()) {
+               udpstream.pacing_rate = ~0U;
+       } else {
+               udpstream.pacing_rate = atoi(pacing_rate_it->second.c_str()) * 1024 / 8;
+       }
+

        config->udpstreams.push_back(udpstream);
        return true;
 }
        config->udpstreams.push_back(udpstream);
        return true;
 }

+

 bool parse_error_log(const ConfigLine &line, Config *config)
 {
        if (line.arguments.size() != 0) {
 bool parse_error_log(const ConfigLine &line, Config *config)
 {
        if (line.arguments.size() != 0) {

diff --git a/config.h b/config.h
index e7959c5d03a4f994f19fb8db4d75e03986a62edb..b84b957c10a6178c32088b69029e0042e229aa09 100644 (file)
--- a/config.h
+++ b/config.h
@@ -18,6 +18,7 @@ struct StreamConfig {
        std::string src;  // Can be empty.
        size_t backlog_size;
        int mark_pool;  // -1 for none.
        std::string src;  // Can be empty.
        size_t backlog_size;
        int mark_pool;  // -1 for none.

+       uint32_t pacing_rate;  // In bytes per second. Default is ~0U (no limit).

        enum { STREAM_ENCODING_RAW = 0, STREAM_ENCODING_METACUBE } encoding;
 };
 
        enum { STREAM_ENCODING_RAW = 0, STREAM_ENCODING_METACUBE } encoding;
 };
 


@@ -25,6 +26,7 @@ struct UDPStreamConfig {
        sockaddr_in6 dst;
        std::string src;  // Can be empty.
        int mark_pool;  // -1 for none.
        sockaddr_in6 dst;
        std::string src;  // Can be empty.
        int mark_pool;  // -1 for none.

+       uint32_t pacing_rate;  // In bytes per second. Default is ~0U (no limit).

 };
 
 struct AcceptorConfig {
 };
 
 struct AcceptorConfig {

diff --git a/cubemap.1 b/cubemap.1
index 9760a895d199977c30c91a8fd8bf3a420697265f..e7c76e37d2d77ec25e5c8824eb33c6d3e712bd3a 100644 (file)
--- a/cubemap.1
+++ b/cubemap.1
@@ -29,6 +29,10 @@ Cubemap also survives the encoder dying and reconnecting.
 .IP \[bu]
 Per-stream fwmark support, for TCP pacing through tc (separate config needed).
 .IP \[bu]
 .IP \[bu]
 Per-stream fwmark support, for TCP pacing through tc (separate config needed).
 .IP \[bu]

+Support for setting max pacing rate through the fq packet scheduler
+(obsoletes the previous point, but depends on experimental kernel patches
+that will hit Linux in 3.13 at the earliest)
+.IP \[bu]

 Reflects anything VLC can reflect over HTTP, even the muxes VLC
 has problems reflecting itself (in particular, FLV).
 .IP \[bu]
 Reflects anything VLC can reflect over HTTP, even the muxes VLC
 has problems reflecting itself (in particular, FLV).
 .IP \[bu]

diff --git a/cubemap.config.sample b/cubemap.config.sample
index c6ab1823a8dbd89faa72f4062b402a29da5bbcac..bce0659325dc400d369b75947bf5903bb3ed2ef0 100644 (file)
--- a/cubemap.config.sample
+++ b/cubemap.config.sample
@@ -33,6 +33,6 @@ error_log type=console
 #
 stream /test.flv src=http://gruessi.zrh.sesse.net:4013/test.flv mark=1000-5000
 stream /test.flv.metacube src=http://gruessi.zrh.sesse.net:4013/test.flv encoding=metacube
 #
 stream /test.flv src=http://gruessi.zrh.sesse.net:4013/test.flv mark=1000-5000
 stream /test.flv.metacube src=http://gruessi.zrh.sesse.net:4013/test.flv encoding=metacube

-stream /udp.ts src=udp://@:1234 backlog_size=1048576
+stream /udp.ts src=udp://@:1234 backlog_size=1048576 pacing_rate_kbit=2000

 udpstream [2001:67c:29f4::50]:5000 src=http://pannekake.samfundet.no:9094/frikanalen.ts.metacube
 udpstream 193.35.52.50:5001 src=http://pannekake.samfundet.no:9094/frikanalen.ts.metacube
 udpstream [2001:67c:29f4::50]:5000 src=http://pannekake.samfundet.no:9094/frikanalen.ts.metacube
 udpstream 193.35.52.50:5001 src=http://pannekake.samfundet.no:9094/frikanalen.ts.metacube

diff --git a/main.cpp b/main.cpp
index c4c5bd4187d3a74850a721fb2d8b90f77835405d..86ba28a81cf41aef2927659375a169f16f03f434 100644 (file)
--- a/main.cpp
+++ b/main.cpp
@@ -184,6 +184,8 @@ void create_streams(const Config &config,
                        servers->set_mark_pool(stream_index, mark_pools[stream_config.mark_pool]);
                }
 
                        servers->set_mark_pool(stream_index, mark_pools[stream_config.mark_pool]);
                }
 

+               servers->set_pacing_rate(stream_index, stream_config.pacing_rate);
+

                string src = stream_config.src;
                if (!src.empty()) {
                        multimap<string, InputWithRefcount>::iterator input_it = inputs->find(src);
                string src = stream_config.src;
                if (!src.empty()) {
                        multimap<string, InputWithRefcount>::iterator input_it = inputs->find(src);


@@ -212,7 +214,7 @@ void create_streams(const Config &config,
                if (udpstream_config.mark_pool != -1) {
                        mark_pool = mark_pools[udpstream_config.mark_pool];
                }
                if (udpstream_config.mark_pool != -1) {
                        mark_pool = mark_pools[udpstream_config.mark_pool];
                }

-               int stream_index = servers->add_udpstream(udpstream_config.dst, mark_pool);
+               int stream_index = servers->add_udpstream(udpstream_config.dst, mark_pool, udpstream_config.pacing_rate);

 
                string src = udpstream_config.src;
                if (!src.empty()) {
 
                string src = udpstream_config.src;
                if (!src.empty()) {

diff --git a/server.cpp b/server.cpp
index 0ed44fe55dbb84c5bb77de14d9c63c1207463d1e..288224cc287dff9480b5443fd83453a9bba6d88b 100644 (file)
--- a/server.cpp
+++ b/server.cpp
@@ -28,6 +28,10 @@
 #include "stream.h"
 #include "util.h"
 
 #include "stream.h"
 #include "util.h"
 

+#ifndef SO_MAX_PACING_RATE
+#define SO_MAX_PACING_RATE 47
+#endif
+

 using namespace std;
 
 extern AccessLogThread *access_log;
 using namespace std;
 
 extern AccessLogThread *access_log;


@@ -271,6 +275,14 @@ void Server::set_mark_pool(int stream_index, MarkPool *mark_pool)
        streams[stream_index]->mark_pool = mark_pool;
 }
 
        streams[stream_index]->mark_pool = mark_pool;
 }
 

+void Server::set_pacing_rate(int stream_index, uint32_t pacing_rate)
+{
+       MutexLock lock(&mutex);
+       assert(clients.empty());
+       assert(stream_index >= 0 && stream_index < ssize_t(streams.size()));
+       streams[stream_index]->pacing_rate = pacing_rate;
+}
+

 void Server::add_data_deferred(int stream_index, const char *data, size_t bytes, StreamStartSuitability suitable_for_stream_start)
 {
        assert(stream_index >= 0 && stream_index < ssize_t(streams.size()));
 void Server::add_data_deferred(int stream_index, const char *data, size_t bytes, StreamStartSuitability suitable_for_stream_start)
 {
        assert(stream_index >= 0 && stream_index < ssize_t(streams.size()));


@@ -525,6 +537,11 @@ int Server::parse_request(Client *client)
                        log_perror("setsockopt(SO_MARK)");
                }
        }
                        log_perror("setsockopt(SO_MARK)");
                }
        }

+       if (setsockopt(client->sock, SOL_SOCKET, SO_MAX_PACING_RATE, &client->stream->pacing_rate, sizeof(client->stream->pacing_rate)) == -1) {
+               if (client->stream->pacing_rate != 0) {
+                       log_perror("setsockopt(SO_MAX_PACING_RATE)");
+               }
+       }

        client->request.clear();
 
        return 200;  // OK!
        client->request.clear();
 
        return 200;  // OK!

diff --git a/server.h b/server.h
index 4b3aab8e9b7250eb47e634805f1978c5a4071ce2..84f822d6fa629e66fce7149d6e1c724e10e53291 100644 (file)
--- a/server.h
+++ b/server.h
@@ -43,6 +43,10 @@ public:
        // NOTE: This should be set before any clients are connected!
        void set_mark_pool(int stream_index, MarkPool *mark_pool);
 
        // NOTE: This should be set before any clients are connected!
        void set_mark_pool(int stream_index, MarkPool *mark_pool);
 

+       // Set that the given stream should use the given max pacing rate from now on.
+       // NOTE: This should be set before any clients are connected!
+       void set_pacing_rate(int stream_index, uint32_t pacing_rate);
+

        // These will be deferred until the next time an iteration in do_work() happens,
        // and the order between them are undefined.
        // XXX: header should ideally be ordered with respect to data.
        // These will be deferred until the next time an iteration in do_work() happens,
        // and the order between them are undefined.
        // XXX: header should ideally be ordered with respect to data.

diff --git a/serverpool.cpp b/serverpool.cpp
index 936bd21293ab074504965dc0f035fbe6a8dc98a0..1c99d4065ef50a39fe808ebcb72331a81d8e0856 100644 (file)
--- a/serverpool.cpp
+++ b/serverpool.cpp
@@ -121,9 +121,9 @@ int ServerPool::add_stream_from_serialized(const StreamProto &stream, const vect
        return num_http_streams++;
 }
        
        return num_http_streams++;
 }
        

-int ServerPool::add_udpstream(const sockaddr_in6 &dst, MarkPool *mark_pool)
+int ServerPool::add_udpstream(const sockaddr_in6 &dst, MarkPool *mark_pool, int pacing_rate)

 {
 {

-       udp_streams.push_back(new UDPStream(dst, mark_pool));
+       udp_streams.push_back(new UDPStream(dst, mark_pool, pacing_rate));

        return num_http_streams + udp_streams.size() - 1;
 }
 
        return num_http_streams + udp_streams.size() - 1;
 }
 


@@ -191,6 +191,13 @@ void ServerPool::set_mark_pool(int stream_index, MarkPool *mark_pool)
 {
        for (int i = 0; i < num_servers; ++i) {
                servers[i].set_mark_pool(stream_index, mark_pool);
 {
        for (int i = 0; i < num_servers; ++i) {
                servers[i].set_mark_pool(stream_index, mark_pool);

+       }
+}
+
+void ServerPool::set_pacing_rate(int stream_index, uint32_t pacing_rate)
+{
+       for (int i = 0; i < num_servers; ++i) {
+               servers[i].set_pacing_rate(stream_index, pacing_rate);

        }       
 }
 
        }       
 }
 

diff --git a/serverpool.h b/serverpool.h
index b67fa48159f8ca7f641097494284fe95e481ccfa..3db9dcce3740ba754266893ffab035efb24c9a29 100644 (file)
--- a/serverpool.h
+++ b/serverpool.h
@@ -33,7 +33,7 @@ public:
        int add_stream(const std::string &url, size_t backlog_size, Stream::Encoding encoding);
        int add_stream_from_serialized(const StreamProto &stream, const std::vector<int> &data_fds);
        void delete_stream(const std::string &url);
        int add_stream(const std::string &url, size_t backlog_size, Stream::Encoding encoding);
        int add_stream_from_serialized(const StreamProto &stream, const std::vector<int> &data_fds);
        void delete_stream(const std::string &url);

-       int add_udpstream(const sockaddr_in6 &dst, MarkPool *mark_pool);
+       int add_udpstream(const sockaddr_in6 &dst, MarkPool *mark_pool, int pacing_rate);

 
        // Returns the stream index for the given URL (e.g. /foo.ts). Returns -1 on failure.
        int lookup_stream_by_url(const std::string &url) const;
 
        // Returns the stream index for the given URL (e.g. /foo.ts). Returns -1 on failure.
        int lookup_stream_by_url(const std::string &url) const;


@@ -47,6 +47,9 @@ public:
        // Connects the given stream to the given mark pool for all the servers.
        void set_mark_pool(int stream_index, MarkPool *mark_pool);
 
        // Connects the given stream to the given mark pool for all the servers.
        void set_mark_pool(int stream_index, MarkPool *mark_pool);
 

+       // Sets the max pacing rate for all the servers.
+       void set_pacing_rate(int stream_index, uint32_t pacing_rate);
+

        // Changes the given stream's backlog size on all the servers.
        void set_backlog_size(int stream_index, size_t new_size);
 
        // Changes the given stream's backlog size on all the servers.
        void set_backlog_size(int stream_index, size_t new_size);
 

diff --git a/stream.cpp b/stream.cpp
index b7a92c2293832e7d77c3b0c8b4c20e604b195a44..6bb86c41e0fc8332bfe4f08a207708d3613fef45 100644 (file)
--- a/stream.cpp
+++ b/stream.cpp
@@ -26,6 +26,7 @@ Stream::Stream(const string &url, size_t backlog_size, Encoding encoding)
          bytes_received(0),
          last_suitable_starting_point(-1),
          mark_pool(NULL),
          bytes_received(0),
          last_suitable_starting_point(-1),
          mark_pool(NULL),

+         pacing_rate(~0U),

          queued_data_last_starting_point(-1)
 {
        if (data_fd == -1) {
          queued_data_last_starting_point(-1)
 {
        if (data_fd == -1) {


@@ -51,6 +52,7 @@ Stream::Stream(const StreamProto &serialized, int data_fd)
          backlog_size(serialized.backlog_size()),
          bytes_received(serialized.bytes_received()),
          mark_pool(NULL),
          backlog_size(serialized.backlog_size()),
          bytes_received(serialized.bytes_received()),
          mark_pool(NULL),

+         pacing_rate(~0U),

          queued_data_last_starting_point(-1)
 {
        if (data_fd == -1) {
          queued_data_last_starting_point(-1)
 {
        if (data_fd == -1) {

diff --git a/stream.h b/stream.h
index 2cdba11ad46b48a1d07801dccd7821efa11568ce..d29b9246a06f76a776385bad4290f69b6cb171be 100644 (file)
--- a/stream.h
+++ b/stream.h
@@ -84,6 +84,9 @@ struct Stream {
        // What pool to fetch marks from, or NULL.
        MarkPool *mark_pool;
 
        // What pool to fetch marks from, or NULL.
        MarkPool *mark_pool;
 

+       // Maximum pacing rate for the stream.
+       int pacing_rate;
+

        // Queued data, if any. Protected by <queued_data_mutex>.
        // The data pointers in the iovec are owned by us.
        std::vector<iovec> queued_data;
        // Queued data, if any. Protected by <queued_data_mutex>.
        // The data pointers in the iovec are owned by us.
        std::vector<iovec> queued_data;

diff --git a/udpstream.cpp b/udpstream.cpp
index 5ed16df2f177d22a72ff008937b5e68f0d45743a..11e152746111568407d7ef5c6714eede9f63a217 100644 (file)
--- a/udpstream.cpp
+++ b/udpstream.cpp
@@ -6,10 +6,15 @@
 #include "udpstream.h"
 #include "util.h"
 
 #include "udpstream.h"
 #include "util.h"
 

-UDPStream::UDPStream(const sockaddr_in6 &dst, MarkPool *mark_pool)
+#ifndef SO_MAX_PACING_RATE
+#define SO_MAX_PACING_RATE 47
+#endif
+
+UDPStream::UDPStream(const sockaddr_in6 &dst, MarkPool *mark_pool, uint32_t pacing_rate)

        : dst(dst),
          mark_pool(mark_pool),
        : dst(dst),
          mark_pool(mark_pool),

-         fwmark(0)
+         fwmark(0),
+         pacing_rate(pacing_rate)

 {
        sock = socket(AF_INET6, SOCK_DGRAM, IPPROTO_UDP);
        if (sock == -1) {
 {
        sock = socket(AF_INET6, SOCK_DGRAM, IPPROTO_UDP);
        if (sock == -1) {


@@ -26,6 +31,11 @@ UDPStream::UDPStream(const sockaddr_in6 &dst, MarkPool *mark_pool)
                        }
                }
        }
                        }
                }
        }

+       if (setsockopt(sock, SOL_SOCKET, SO_MAX_PACING_RATE, &pacing_rate, sizeof(pacing_rate)) == -1) {
+               if (pacing_rate != 0) {
+                       log_perror("setsockopt(SO_MAX_PACING_RATE)");
+               }
+       }

 }
 
 UDPStream::~UDPStream()
 }
 
 UDPStream::~UDPStream()

diff --git a/udpstream.h b/udpstream.h
index e30c9481a365e3d0e8d3700f0c44c2cf0b413bc4..4d1e1f7ba39c693fc6344136ac01e9d4cad1b0ad 100644 (file)
--- a/udpstream.h
+++ b/udpstream.h
@@ -17,7 +17,7 @@ class MarkPool;
 class UDPStream {
 public:
        // <mark_pool> can be NULL. Does not take ownership of the mark pool.
 class UDPStream {
 public:
        // <mark_pool> can be NULL. Does not take ownership of the mark pool.

-       UDPStream(const sockaddr_in6 &dst, MarkPool *mark_pool);
+       UDPStream(const sockaddr_in6 &dst, MarkPool *mark_pool, uint32_t pacing_rate);

        ~UDPStream();
 
        void send(const char *data, size_t bytes);
        ~UDPStream();
 
        void send(const char *data, size_t bytes);


@@ -29,6 +29,7 @@ private:
        int sock;
        MarkPool *mark_pool;
        int fwmark;
        int sock;
        MarkPool *mark_pool;
        int fwmark;

+       uint32_t pacing_rate;

 };
 
 #endif  // !defined(_UDPSTREAM_H)
 };
 
 #endif  // !defined(_UDPSTREAM_H)