UDP File Sender&Receiver

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.

The sender application

The example code for the sender is shown below.

// 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 receiver application

The example code for the receiver is shown below.

// 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;
}