void runner(std::size_t thread_index, boost::barrier& data_barrier, data_t& data) {
    for (std::size_t i = 0; i < 1000; ++ i) {
        fill_data(data.at(thread_index));
        data_barrier.wait();

        if (!thread_index) {
            compute_send_data(data);
        }
        data_barrier.wait();
    }
}

 std::pair<std::vector<unsigned> , unsigned> evaluate(unsigned num_iterations)
 {
     thread_num_iterations = num_iterations;
     thread_score = std::vector<unsigned>(def_decks.size(), 0u);
     thread_total = 0;
     thread_compare = false;
     // unlock all the threads
     main_barrier.wait();
     // wait for the threads
     main_barrier.wait();
     return(std::make_pair(thread_score, thread_total));
 }

 bool start() {
   bool throws = false;
   t_ = std::thread([&]() {
     server_->start([&]() { barrier_.wait(); },
                    [&](std::exception_ptr /*ex*/) {
                      throws = true;
                      server_ = nullptr;
                      barrier_.wait();
                    });
   });
   barrier_.wait();
   return !throws;
 }

 std::pair<std::vector<unsigned> , unsigned> compare(unsigned num_iterations, double prev_score)
 {
     thread_num_iterations = num_iterations;
     thread_score = std::vector<unsigned>(def_decks.size(), 0u);
     thread_total = 0;
     thread_prev_score = prev_score;
     thread_compare = true;
     thread_compare_stop = false;
     // unlock all the threads
     main_barrier.wait();
     // wait for the threads
     main_barrier.wait();
     return(std::make_pair(thread_score, thread_total));
 }

int main(int argc, char* argv[])
{
    if (argc < 2)
    {
        boost::filesystem::path exePath(argv[0]);
        std::cout << "usage: " << exePath.filename().string() << " <RTSP url of stream>" << std::endl;
        return -1;
    }
    
    boost::program_options::options_description desc("");
	desc.add_options()
		("no-display", "")
		("url", boost::program_options::value<std::string>(), "url")
		("interface", boost::program_options::value<std::string>(), "interface")
		("buffer-size", boost::program_options::value<unsigned long>(), "buffer-size");
    
    boost::program_options::positional_options_description p;
    p.add("url", -1);
    
    boost::program_options::variables_map vm;
    boost::program_options::store(boost::program_options::command_line_parser(argc, argv).
              options(desc).positional(p).run(), vm);
    boost::program_options::notify(vm);
    
    const char* interfaceAddress;
    if (vm.count("interface"))
    {
		interfaceAddress = vm["interface"].as<std::string>().c_str();
    }
    else
    {
		interfaceAddress = "0.0.0.0";
    }

	unsigned long bufferSize;
	if (vm.count("buffer-size"))
	{
		bufferSize = vm["buffer-size"].as<unsigned long>();
	}
	else
	{
		bufferSize = DEFAULT_SINK_BUFFER_SIZE;
	}

	std::cout << "Buffer size " << to_human_readable_byte_count(bufferSize, false, false) << std::endl;

	rtspClient = RTSPCubemapSourceClient::create(vm["url"].as<std::string>().c_str(), bufferSize, AV_PIX_FMT_RGBA, false, interfaceAddress);
    std::function<void (RTSPCubemapSourceClient*, CubemapSource*)> callback(boost::bind(&onDidConnect, _1, _2));
    rtspClient->setOnDidConnect(callback);
    rtspClient->connect();
    
    barrier.wait();
    
    Renderer renderer(cubemapSource);
	renderer.setOnDisplayedCubemapFace(boost::bind(&onDisplayedCubemapFace, _1, _2));
	renderer.setOnDisplayedFrame(boost::bind(&onDisplayedFrame, _1));
    renderer.start(); // Returns when window is closed
    
    CubemapSource::destroy(cubemapSource);
}

 ~Process()
 {
     destroy_threads = true;
     main_barrier.wait();
     for(auto thread: threads) { thread->join(); }
     for(auto data: threads_data) { delete(data); }
 }

void wait_fn( boost::barrier & b,
              boost::fibers::mutex & mtx,
              boost::fibers::condition_variable & cond,
              bool & flag) {
    b.wait();
	std::unique_lock< boost::fibers::mutex > lk( mtx);
	cond.wait( lk, [&flag](){ return flag; });
	++value;
}

void invoke_n_workers(
    boost::barrier& b
    , double& elapsed
    , boost::uint64_t workers
)
{
    b.wait();

    invoke_n_workers_nowait(elapsed, workers);
}

static void
doit(const std::vector<const FgVariant*> & /*sources*/,
     const std::vector<FgVariant*> & /*sinks*/)
{
        // First wait for both threads to reach so we can force
        // one exception to be thrown first. Note that if you get
        // a deadlock here, something is seriously screwed :-)
    m_barrier.wait();
    fgThrow("Thrown from multiple threads at the same time!");
}

void notify_all_fn( boost::barrier & b,
                    boost::fibers::mutex & mtx,
                    boost::fibers::condition_variable & cond,
                    bool & flag) {
    b.wait();
	std::unique_lock< boost::fibers::mutex > lk( mtx);
    flag = true;
    lk.unlock();
	cond.notify_all();
}

static void thread_func_TestPercentiles(boost::barrier& bar, MetricsFactory* factory, const char* name)
{
    Metric& metric = factory->get(name);
    bar.wait();
    for (int j=0; j < 25; j++) {
        double source[] = { 43, 54, 56, 61, 62, 66, 68, 69, 69, 70, 71, 72, 77, 78, 79, 85, 87, 88, 89, 93, 95, 96, 98, 99, 99 };
        for (size_t i=0; i < sizeof(source)/sizeof(*source); i++) {
            metric.update(source[i]);
        }
    }
}

//! This function is executed in multiple threads
void thread_fun(boost::barrier& bar)
{
    // Wait until all threads are created
    bar.wait();

    // Now, do some logging
    for (unsigned int i = 0; i < LOG_RECORDS_TO_WRITE; ++i)
    {
        BOOST_LOG(test_lg::get()) << "Log record " << i;
    }
}

void test(unsigned int record_count, boost::barrier& bar)
{
    BOOST_LOG_SCOPED_THREAD_TAG("ThreadID", boost::this_thread::get_id());
    bar.wait();

    src::severity_logger< severity_level > slg;
    for (unsigned int i = 0; i < record_count; ++i)
    {
        BOOST_LOG_SEV(slg, warning) << "Test record";
    }
}

static void thread_func_TestHistogram(boost::barrier& bar, MetricsFactory* factory, const char* name)
{
    Metric& metric = factory->get(name);
    bar.wait();
    for (int j=0; j < 100; j++) {
        metric.update(0);
        metric.update(1);
        metric.update(10);
        metric.update(30);
        metric.update(31);
    }
}

void threadFunc(FreeList& pool, ros::atomic<bool>& done, ros::atomic<bool>& failed, boost::barrier& b)
{
  b.wait();

  //ROS_INFO_STREAM("Thread " << boost::this_thread::get_id() << " starting");

  uint32_t* vals[10];
  uint64_t alloc_count = 0;
  while (!done.load())
  {
    for (uint32_t i = 0; i < 10; ++i)
    {
      vals[i] = static_cast<uint32_t*>(pool.allocate());

      if (vals[i])
      {
        ++alloc_count;
        *vals[i] = i;
      }
      else
      {
        ROS_ERROR_STREAM("Thread " << boost::this_thread::get_id() << " failed to allocate");
      }
    }

    for (uint32_t i = 0; i < 10; ++i)
    {
      if (vals[i])
      {
        if (*vals[i] != i)
        {
          ROS_ERROR_STREAM("Thread " << boost::this_thread::get_id() << " val " << vals[i] << " " << i << " = " << *vals[i]);
          failed.store(true);
        }

        pool.free(vals[i]);
      }
    }

    if (failed.load())
    {
#if FREE_LIST_DEBUG
      boost::mutex::scoped_lock lock(g_debug_mutex);
      g_debug.push_back(*pool.getDebug());
#endif
      return;
    }
  }

  //ROS_INFO_STREAM("Thread " << boost::this_thread::get_id() << " allocated " << alloc_count << " blocks");
}

//! This function is executed in multiple threads
void thread_fun(boost::barrier& bar)
{
    // Wait until all threads are created
    bar.wait();

    // Here we go. First, identify the thread.
    BOOST_LOG_SCOPED_THREAD_TAG("ThreadID", boost::this_thread::get_id());

    // Now, do some logging
    for (unsigned int i = 0; i < LOG_RECORDS_TO_WRITE; ++i)
    {
        BOOST_LOG(test_lg::get()) << "Log record " << i;
    }
}

inline void worker(
    boost::barrier& b
  , boost::uint64_t updates
    )
{
    b.wait();

    for (double i = 0.; i < updates; ++i)
    {
        global_scratch.reset(new double);

        *global_scratch += 1. / (2. * i * (*global_scratch) + 1.);

        global_scratch.reset();
    }
}

inline void worker(
    boost::barrier& b
  , std::uint64_t updates
    )
{
    b.wait();

    for (double i = 0.; i < updates; ++i)
    {
        global_scratch = new double;

        *global_scratch += 1. / (2. * i * (*global_scratch) + 1.);

        delete global_scratch;
    }
}

    void operator() () {
        timestamp::buf_type buf[2];

        if (barrier != NULL)
            barrier->wait();

        for (int i=0; i < iterations; i++) {
            hr.start_incr();
            int n = timestamp::update_and_write(
                DATE_TIME_WITH_USEC, buf[0], sizeof(buf[0]));
            hr.stop_incr();
            time_val t1 = now_utc();
            if (n != 24 || strlen(buf[0]) != 24) {
                std::cerr << "Wrong buffer length: " << buf[0] << std::endl;
                BOOST_REQUIRE_EQUAL(24, n);
            }
            hr.start_incr();
            timestamp::update_and_write(
                DATE_TIME_WITH_USEC, buf[1], sizeof(buf[1]));
            hr.stop_incr();
            time_val t2 = now_utc();
            if (strcmp(buf[0], buf[1]) > 0) {
                std::cerr << "Backward time jump detected: "
                    << buf[0] << ' ' << buf[1]
                    << '(' << t1.sec() << ' ' << t2.sec() << ')' << std::endl;

                BOOST_REQUIRE(strcmp(buf[1], buf[0]) >= 0);
            }
        }

        const time_val& tv = hr.elapsed_time();
        timestamp::format(TIME_WITH_USEC, tv, buf[0], sizeof(buf[0]));

        {
            std::stringstream s; s.precision(6);
            s << "Thread" << id << " timestamp: "
              #ifdef DEBUG_TIMESTAMP
              << "hrcalls=" << timestamp::hrcalls()
              << ", syscalls=" << timestamp::syscalls()
              #endif
              << ", speed=" << std::fixed << ((double)iterations / tv.seconds())
              << ", latency=" << (1000000000.0*tv.seconds()/iterations) << " ns";
            BOOST_TEST_MESSAGE(s.str());
        }
    }

void onDidConnect(RTSPCubemapSourceClient* client, CubemapSource* cubemapSource)
{
    if (typeid(cubemapSource) == typeid(H264CubemapSource))
    {
        H264CubemapSource* h264CubemapSource = (H264CubemapSource*)cubemapSource;
        
        h264CubemapSource->setOnReceivedNALU       (boost::bind(&onReceivedNALU,        _1, _2, _3, _4));
        h264CubemapSource->setOnReceivedFrame      (boost::bind(&onReceivedFrame,       _1, _2, _3, _4));
        h264CubemapSource->setOnDecodedFrame       (boost::bind(&onDecodedFrame,        _1, _2, _3, _4));
        h264CubemapSource->setOnColorConvertedFrame(boost::bind(&onColorConvertedFrame, _1, _2, _3, _4));
    }
    
    stats.autoSummary(boost::chrono::seconds(10),
					  AlloReceiver::statValsMaker,
					  AlloReceiver::postProcessorMaker,
					  AlloReceiver::formatStringMaker);
    
    ::cubemapSource = cubemapSource;
    
    barrier.wait();
}