This pair of examples shows how to encode and transmit a file over a UDP socket, then receive and decode that file at the other end. The sender and receiver applications are standalone, and you can run them on the same computer (localhost is the default target address) or over an actual IP network.
The receiver should be started before the sender, because the sender starts by transmitting the metadata that describes the file parameters. If the receiver does not get this metadata packet, then it cannot receive the file. In a realistic scenario, the sender can be modified to transmit this metadata periodically or upon request when a new receiver joins.
The input file is automatically split into multiple blocks if necessary, and you can control when the sender should jump from one block to the next. The sender will continuously transmit data packets at the specified data rate until you stop the application.
Table of Contents
The example code for the sender is shown below.
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 | // Copyright Steinwurf ApS 2017.
// Distributed under the "STEINWURF EVALUATION LICENSE 1.0".
// See accompanying file LICENSE.rst or
// http://www.steinwurf.com/licensing
#include <algorithm>
#include <cassert>
#include <cstdint>
#include <cstdlib>
#include <ctime>
#include <iostream>
#include <fstream>
#include <string>
#include <thread>
#include <vector>
#include <boost/asio.hpp>
#include <boost/chrono.hpp>
#include <boost/program_options.hpp>
#include <kodo_core/object/interleaved_file_encoder.hpp>
#include <kodo_rlnc/coders.hpp>
#include "compute_sha1.hpp"
/// @example udp_file_sender.cpp
///
/// Simple example showing how to encode and send a file over a UDP socket
int main(int argc, char* argv[])
{
namespace bpo = boost::program_options;
namespace bc = boost::chrono;
// Set the number of symbols and the size of a symbol in bytes
uint32_t symbols;
uint32_t symbol_size;
uint32_t interleaving_batch;
// UDP endpoint (where the receiver will be listening)
std::string remote_ip;
uint16_t remote_port;
uint32_t data_rate;
std::string filename;
uint32_t random_bytes;
// Parse the program options
bpo::options_description options("File sender options");
options.add_options()
("remote_ip,i",
bpo::value<std::string>(&remote_ip)->default_value("127.0.0.1"),
"Specify the remote IPv4 address to which we send the data")
("remote_port,p", bpo::value<uint16_t>(&remote_port)->default_value(45678),
"Set the remote port where the receiver is listening")
("input_file,f",
bpo::value<std::string>(&filename)->default_value("encode-test.bin"),
"Path to the file that should be sent (the file will be overwritten if "
"random_bytes is specified)")
("random_bytes,r",
bpo::value<uint32_t>(&random_bytes),
"Create a file of the specified size and fill it with random data")
("symbols,s", bpo::value<uint32_t>(&symbols)->default_value(100),
"Set the maximum number of symbols in a block")
("symbol_size,z", bpo::value<uint32_t>(&symbol_size)->default_value(1400),
"Set the size of a symbols in bytes")
("interleaving_batch,b",
bpo::value<uint32_t>(&interleaving_batch)->default_value(10),
"The number of symbols sent for each block before jumping to the next")
("data_rate,d",
bpo::value<uint32_t>(&data_rate)->default_value(1000000),
"The desired outgoing data rate in bytes/second")
("help,h", "Print this help message");
bpo::variables_map opts;
// Verify all required options are present
try
{
bpo::store(bpo::parse_command_line(argc, argv, options), opts);
if (opts.count("help"))
{
std::cout << options << std::endl;
return 0;
}
bpo::notify(opts);
}
catch (const std::exception& e)
{
std::cout << "Error when parsing command-line options: " << e.what()
<< std::endl;
std::cout << "See list of options with \"" << argv[0] << " --help\""
<< std::endl;
return 0;
}
// We will generate random data to transmit
srand((uint32_t)time(0));
// Typdefs for the encoder/decoder type we wish to use
fifi::api::field field = fifi::api::field::binary;
using rlnc_encoder = kodo_rlnc::encoder;
using file_encoder =
kodo_core::object::interleaved_file_encoder<rlnc_encoder>;
// In the following we will make an encoder/decoder factory.
// The factories are used to build actual encoders/decoders
file_encoder::factory encoder_factory(field, symbols, symbol_size);
// If random_bytes is specified, we will overwrite the file contents
// with random data
if (opts.count("random_bytes") && random_bytes > 0)
{
std::cout << "Writing " << random_bytes << " random bytes to file: "
<< filename << std::endl;
// Make sure that the data file is not present from a previous run
std::remove(filename.c_str());
// Fill the data buffer with random bytes
std::vector<uint8_t> data(random_bytes);
std::generate(data.begin(), data.end(), rand);
std::string sha1 = compute_sha1(data.data(), (uint32_t) data.size());
std::cout << "SHA1 checksum: " << sha1 << std::endl;
// Create a test file that will contain the random bytes
std::ofstream encode_file;
encode_file.open(filename, std::ofstream::binary);
encode_file.write((char*)data.data(), data.size());
encode_file.close();
}
encoder_factory.set_filename(filename);
auto encoder = encoder_factory.build();
uint32_t object_size = encoder->object_size();
std::cout << "Encoder parameters: " << std::endl;
std::cout << " Object size: " << object_size << std::endl;
std::cout << " Symbols: " << symbols << std::endl;
std::cout << " Symbol size: " << symbol_size << std::endl;
// The endpoint (represents the reciever)
auto receiver = boost::asio::ip::udp::endpoint(
boost::asio::ip::address::from_string(remote_ip), remote_port);
// Create a boost::asio socket
boost::asio::io_service io_service;
boost::asio::ip::udp::socket socket(io_service);
socket.open(boost::asio::ip::udp::v4());
// The first packet will contain the metadata necessary for the decoder
std::vector<uint8_t> metadata(16);
uint8_t* buffer = metadata.data();
// The packet starts with the 4-byte magic string "META"
std::string meta("META");
meta.copy((char*)buffer, 4, 0);
buffer += 4;
endian::big_endian::put<uint32_t>(object_size, buffer);
buffer += sizeof(uint32_t);
endian::big_endian::put<uint32_t>(symbols, buffer);
buffer += sizeof(uint32_t);
endian::big_endian::put<uint32_t>(symbol_size, buffer);
buffer += sizeof(uint32_t);
// In a practical system, the metadata might be transmitted periodically
// to allow the late-joining receivers to complete the file transfer.
// Alternatively, the receivers might acquire the metadata by pulling this
// information from the sender using a separate unicast connection.
// In this example, we simply start the transmission with the metadata.
socket.send_to(boost::asio::buffer(metadata), receiver);
std::cout << "Metadata sent." << std::endl;
uint32_t total_bytes = 0;
uint32_t packets_sent = 0;
bc::high_resolution_clock::time_point start, stop, now;
bool running = true;
std::thread send_thread([&]()
{
start = bc::high_resolution_clock::now();
uint32_t current_batch = 0;
std::vector<uint8_t> payload;
while (running)
{
// Stall the sending thread to match the desired data rate
while (true)
{
now = bc::high_resolution_clock::now();
double elapsed_time =
(double) bc::duration_cast<bc::microseconds>(
now - start).count();
double current_rate = total_bytes / elapsed_time;
// Exit the while loop if the current rate drops below the
// desired rate
if (current_rate * 1000000 <= data_rate)
break;
}
// Send a payload for the current block
uint32_t b = encoder->current_block();
payload.resize(encoder->payload_size(b));
// Write a symbol into the payload buffer
encoder->write_payload(payload.data());
total_bytes +=
socket.send_to(boost::asio::buffer(payload), receiver);
packets_sent++;
current_batch++;
// Advance to the next block if the current batch is finished
if (current_batch == interleaving_batch)
{
encoder->advance_block();
current_batch = 0;
}
}
});
// Wait for a keypress to finish the sending thread
std::cout << "Sender thread is running. Press ENTER to exit!" << std::endl;
std::cin.ignore();
running = false;
send_thread.join();
stop = bc::high_resolution_clock::now();
double total_time =
(double)(bc::duration_cast<bc::microseconds>(stop - start).count());
double sending_rate = total_bytes / total_time;
std::cout << "Total time: " << total_time / 1000000.0 << " s" << std::endl;
std::cout << "Sending rate: " << sending_rate << " MB/s" << std::endl;
}
|
The example code for the receiver is shown below.
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 | // Copyright Steinwurf ApS 2017.
// Distributed under the "STEINWURF EVALUATION LICENSE 1.0".
// See accompanying file LICENSE.rst or
// http://www.steinwurf.com/licensing
#include <algorithm>
#include <cassert>
#include <cstdint>
#include <cstring>
#include <cstdlib>
#include <ctime>
#include <iostream>
#include <fstream>
#include <vector>
#include <string>
#include <boost/asio.hpp>
#include <boost/chrono.hpp>
#include <boost/program_options.hpp>
#include <kodo_core/object/interleaved_file_decoder.hpp>
#include <kodo_rlnc/coders.hpp>
#include "compute_sha1.hpp"
/// @example udp_file_receiver.cpp
///
/// Simple example showing how to receive and decode a file over a UDP socket
int main(int argc, char* argv[])
{
namespace bpo = boost::program_options;
namespace bc = boost::chrono;
// We will generate random numbers to simulate losses
srand((uint32_t)time(0));
// UDP endpoint (where the receiver will be listening)
std::string local_ip;
uint16_t local_port;
std::string filename;
// Percentage of packets that will be dropped by the decoder
// (this is useful for simulation purposes)
uint32_t loss_rate;
// Parse the program options
bpo::options_description options("File receiver options");
options.add_options()
("local_ip,i",
bpo::value<std::string>(&local_ip)->default_value("0.0.0.0"),
"Specify the local IPv4 address of the listening interface")
("local_port,p", bpo::value<uint16_t>(&local_port)->default_value(45678),
"Set the local port where the receiver is listening")
("output_file,f",
bpo::value<std::string>(&filename)->default_value("decode-test.bin"),
"Path to the output file (this file will be overwritten!)")
("loss_rate,l", bpo::value<uint32_t>(&loss_rate)->default_value(0),
"Percentage of received packets that should be dropped[0-100]")
("help,h", "Print this help message");
bpo::variables_map opts;
// Verify all required options are present
try
{
bpo::store(bpo::parse_command_line(argc, argv, options), opts);
if (opts.count("help"))
{
std::cout << options << std::endl;
return 0;
}
bpo::notify(opts);
}
catch (const std::exception& e)
{
std::cout << "Error when parsing command-line options: " << e.what()
<< std::endl;
std::cout << "See list of options with \"" << argv[0] << " --help\""
<< std::endl;
return 0;
}
// Allocate some storage for a payload buffer used to receive data from
// the network. This buffer should be as large as the MTU (or larger).
std::vector<uint8_t> payload(10000);
// Create a boost::asio socket
boost::asio::io_service io_service;
boost::asio::ip::udp::socket socket(io_service);
socket.open(boost::asio::ip::udp::v4());
socket.bind(boost::asio::ip::udp::endpoint(
boost::asio::ip::address::from_string(local_ip), local_port));
std::cout << "Ready to receive metadata..." << std::endl;
boost::asio::ip::udp::endpoint sender;
boost::asio::ip::udp::endpoint original_sender;
uint32_t symbols;
uint32_t symbol_size;
uint32_t object_size;
// The receiver waits for a valid metadata packet from the sender
while (true)
{
size_t bytes =
socket.receive_from(boost::asio::buffer(payload), sender);
if (bytes == 16 && strncmp("META", (char*)payload.data(), 4) == 0)
{
std::cout << "Metadata received from: " << sender << std::endl;
original_sender = sender;
uint8_t* buffer = payload.data() + 4;
// Get the necessary parameters from the metadata
endian::big_endian::get<uint32_t>(object_size, buffer);
buffer += sizeof(uint32_t);
endian::big_endian::get<uint32_t>(symbols, buffer);
buffer += sizeof(uint32_t);
endian::big_endian::get<uint32_t>(symbol_size, buffer);
buffer += sizeof(uint32_t);
std::cout << " Object size: " << object_size << std::endl;
std::cout << " Symbols: " << symbols << std::endl;
std::cout << " Symbol size: " << symbol_size << std::endl;
break;
}
}
// Typdefs for the encoder/decoder type we wish to use
fifi::api::field field = fifi::api::field::binary;
using rlnc_decoder = kodo_rlnc::decoder;
using file_decoder =
kodo_core::object::interleaved_file_decoder<rlnc_decoder>;
// In the following we will make an encoder/decoder factory.
// The factories are used to build actual encoders/decoders
file_decoder::factory decoder_factory(field, symbols, symbol_size);
// Make sure that the data file is not present from a previous run
std::remove(filename.c_str());
// Specify the target filename and size for the decoder
decoder_factory.set_filename(filename);
decoder_factory.set_file_size(object_size);
auto decoder = decoder_factory.build();
// Define various packet counters
uint32_t packets_received = 0;
uint32_t packets_processed = 0;
uint32_t packets_dropped = 0;
uint32_t total_bytes = 0;
bc::high_resolution_clock::time_point start, stop;
start = bc::high_resolution_clock::now();
while (!decoder->is_complete())
{
uint32_t bytes =
socket.receive_from(boost::asio::buffer(payload), sender);
if (sender != original_sender)
{
std::cout << "Ignoring packet from: " << sender << std::endl;
continue;
}
packets_received++;
total_bytes += bytes;
// Simulate random packet losses - useful for benchmarking
if (loss_rate > 0 && uint32_t(rand() % 100) < loss_rate)
{
packets_dropped++;
continue;
}
// Pass that packet to the appropriate decoder
decoder->read_payload(payload.data());
packets_processed++;
}
// The file handles should be closed when the decoder is deleted
decoder.reset();
stop = bc::high_resolution_clock::now();
std::cout << std::endl;
std::cout << "Decoding completed." << std::endl;
std::cout << "Packets processed/received: " << packets_processed
<< "/" << packets_received << " (" << packets_dropped
<< " dropped)" << std::endl;
std::cout << "Bytes received: " << total_bytes << std::endl;
double total_time =
(double)(bc::duration_cast<bc::microseconds>(stop - start).count());
double receiving_rate = total_bytes / total_time;
std::cout << "Total time: " << total_time / 1000000.0 << " s" << std::endl;
std::cout << "Receiving rate: " << receiving_rate << " MB/s" << std::endl;
// Read the decoded data from the output file for verification
std::cout << "Calculating checksum for the output file..." << std::endl;
std::vector<uint8_t> data(object_size);
std::ifstream decode_file;
decode_file.open(filename, std::ifstream::binary);
decode_file.read((char*)data.data(), data.size());
decode_file.close();
std::string sha1 = compute_sha1(data.data(), (uint32_t)data.size());
std::cout << "SHA1 checksum: " << sha1 << std::endl;
}
|