int main(int argc, char** argv) {
    init();
    if (handle_args(argc, argv) == 1)
        return 0;

    int processors = boost::thread::hardware_concurrency();

    ioService.post(boost::bind(read_images));
    ioService.post(boost::bind(assign_workers));
    ioService.post(boost::bind(output));

    boost::asio::io_service::work work(ioService);
    for (int i = 0; i < processors; i++) {
        threadpool.create_thread(boost::bind(&boost::asio::io_service::run, &ioService));
    }
    threadpool.join_all();
    return 0;
}

void subscribeHandler(boost::asio::io_service &ioService, const std::vector<char> &buf)
{
    std::string msg(buf.begin(), buf.end());

    std::cerr << "Message: " << msg << std::endl;

    if( msg == "stop" )
        ioService.stop();
}

 void do_run()
 {
     ioService = new boost::asio::io_service();
     s = new sopmq::node::server(*ioService, 8481);
     
     s->start();
     
     ioService->run();
 }

		void syncLoad(AssetPath path, AssetPromisePtr pr, AssetContentPtr c) {
			Progress::Work w = (Progress::Work)boost::filesystem::file_size(path);
			mngr.prog_.addWork( w );
			// add the loading function to the io_service
			io.post( boost::bind(
				&AssetManager::Loader::asyncLoad,
				this, path, pr, c, w
				)
			);
		}

/*! this is why we went through all this efford. You're in control of the event lopp. huray!!!
 * this version is more suitable for realtime applications
 * */
void the_non_blocking_loop(boost::asio::io_service& io_service)
{
	while (true)
	{
		while (io_service.poll_one());

		// do some other work, e.g. sleep
		std::this_thread::sleep_for(std::chrono::milliseconds(100));
	}
}

// workarround windows that can use posix stream for stdin
static void input_thread(boost::asio::io_service & io_service)
{
	while (!boost::this_thread::interruption_requested() && !std::cin.eof())
	{
		std::string line;
		std::getline(std::cin, line);
		io_service.post([line]{
			input_got_one_line(ansi_utf8(line));
		});
	}
}

void PionScheduler::processServiceWork(boost::asio::io_service& service) {
	while (m_is_running) {
		try {
			service.run();
		} catch (std::exception& e) {
			PION_LOG_ERROR(m_logger, e.what());
		} catch (...) {
			PION_LOG_ERROR(m_logger, "caught unrecognized exception");
		}
	}	
}

    /** Constructor.

        @param log A pointer to a stream to log to, or `nullptr`
        to disable logging.
        
        @param threads The number of threads in the io_service.
    */
    server(std::ostream* log, std::size_t threads)
        : log_(log)
        , sock_(ios_)
        , acceptor_(ios_)
        , work_(ios_)
    {
        thread_.reserve(threads);
        for(std::size_t i = 0; i < threads; ++i)
            thread_.emplace_back(
                [&]{ ios_.run(); });
    }

int main()
{
	av_start(&io_service);

	int port = 24950; // "av" = 0x6176 = 24950
	// 开启 av协议处理

	boost::asio::spawn(io_service, boost::bind(&async_acceptor, _1, port));

	// 无限睡眠，客户端的话就开始写客户端的逻辑吧
	io_service.run();
}

int main() 
{ 
    bp::child c = start_child(); 

    bp::pistream &is = c.get_stdout(); 
    in.assign(is.handle().release()); 

    begin_read(); 
    io_service.run(); 

    c.wait(); 
} 

void
connection_t::connect(boost::asio::io_service& ioservice,
                      const endpoint_t& endpoint,
                      unsigned int connect_timeout)
{
    ioservice.reset();

    if(endpoint.is_unix()) {
        auto s = std::make_shared<boost::asio::local::stream_protocol::socket>(ioservice);
        boost::system::error_code error;
        s->async_connect(boost::asio::local::stream_protocol::endpoint(endpoint.get_path()),
                         std::bind(&local_connection_handler,
                                   std::placeholders::_1,
                                   std::ref(error),
                                   std::ref(ioservice)));

        if(run_with_timeout(ioservice, connect_timeout)) {
            throw std::runtime_error("Connection timed out");
        } else if(error) {
            throw boost::system::system_error(error);
        }

        m_socket = s;
    } else {
        boost::asio::ip::tcp::resolver resolver(ioservice);

        tcp_connector conn = {
            ioservice,
            std::shared_ptr<boost::asio::ip::tcp::socket>(),
            boost::system::error_code(),
            std::vector<boost::asio::ip::tcp::endpoint>()
        };

        resolver.async_resolve(
            boost::asio::ip::tcp::resolver::query(
                endpoint.get_host(),
                boost::lexical_cast<std::string>(endpoint.get_port())
            ),
            std::bind(&tcp_connector::resolve_handler,
                      &conn,
                      std::placeholders::_1,
                      std::placeholders::_2)
        );

        if(run_with_timeout(ioservice, connect_timeout)) {
            throw std::runtime_error("Connection timed out");
        } else if(conn.socket) {
            m_socket = conn.socket;
        } else {
            throw boost::system::system_error(conn.error);
        }
    }
}

int main(int ac, char *av[])
{
    gflags_log_name = const_cast<char *>("nrad6");

    process_options(ac, av);

    io_service.run();

    dynlease_serialize(leasefile);

    std::exit(EXIT_SUCCESS);
}

void controller::run_io_service()
{
    LOG(INFO) << "IO service started.";
    try
    {
        io_service_.run();
    }
    catch (std::exception& e)
    {
        LOG(ERROR) << e.what();
    }
}

// MsgWaitForMultipleObjectsEx 集成进去.
static inline void avloop_run_gui(boost::asio::io_service& io_service)
{
	using namespace ::detail;

	boost::asio::io_service::work work(io_service);

	if (!boost::asio::has_service<IdleService>(io_service))
		boost::asio::add_service(io_service, new IdleService(io_service));

	if (!boost::asio::has_service<Win32MsgLoopService>(io_service))
		boost::asio::add_service(io_service, new Win32MsgLoopService(io_service));

	while (!io_service.stopped())
	{
		// 首先处理 asio 的消息.
		while (io_service.poll())
		{
			// 然后执行 gui 循环，看有没有 gui 事件.
			if (boost::asio::use_service<Win32MsgLoopService>(io_service).has_message())
			{
				// 执行以下.
				boost::asio::use_service<Win32MsgLoopService>(io_service).poll_one();
			}
		}
		// 然后执行 gui 循环，看有没有 gui 事件.
		while(boost::asio::use_service<Win32MsgLoopService>(io_service).has_message())
		{
			// 执行以下.
			boost::asio::use_service<Win32MsgLoopService>(io_service).poll_one();
		}

		// 执行 idle handler!
		if (boost::asio::use_service<IdleService>(io_service).has_idle())
			boost::asio::use_service<IdleService>(io_service).poll_one();

		// 都没有事件了，执行 一次 1ms 的超时等待.
		auto ret = MsgWaitForMultipleObjectsEx(0, nullptr, 1, QS_ALLEVENTS, MWMO_WAITALL|MWMO_ALERTABLE | MWMO_INPUTAVAILABLE);
		// 可能是有 gui 消息了， 呵呵， 从头来吧.
	}
}

int main(int argc, char** argv)
{
	po::options_description desc("Options");
	
	desc.add_options()
		("port,p", po::value<int>(), "Port")
		("fifo_size,F", po::value<int>(), "FIFO size")
		("fifo_low_watermark,L", po::value<unsigned int>(), "FIFO low watermark")
		("fifo_high_watermark,H", po::value<unsigned int>(), "FIFO high watermark")
		("buf_len,X", po::value<unsigned int>(), "Buffer length")
		("tx_interval,i", po::value<int>(), "Mixer call interval");
	
	po::variables_map vm;
	po::store(po::parse_command_line(argc, argv, desc), vm);
	po::notify(vm);
	
	if (vm.count("port"))
		port = vm["port"].as<int>();
	if (vm.count("fifo_size"))
		fifo_size = vm["fifo_size"].as<int>();
		
	fifo_high = fifo_size;
		
	if (vm.count("fifo_low_watermark"))
		fifo_low = vm["fifo_low_watermark"].as<unsigned int>();
	if (vm.count("fifo_high_watermark"))
		fifo_high = vm["fifo_high_watermark"].as<unsigned int>();
	if (vm.count("buf_len"))
		buf_len = vm["buf_len"].as<unsigned int>();
	if (vm.count("tx_interval"))
		tx_interval = vm["tx_interval"].as<int>();
	
	tcp_endpoint = tcp::endpoint(tcp::v6(), port);
	udp_endpoint = udp::endpoint(udp::v6(), port);
	acceptor = tcp::acceptor(io_service, tcp_endpoint);
	udp_socket = udp::socket(io_service, udp_endpoint);
	
	tcp_socket = new tcp::socket(io_service);
	
	acceptor.listen();
	acceptor.async_accept(*tcp_socket, receive_connection);
	udp_socket.async_receive_from(ba::buffer(buffer), client_udp_endpoint, read_udp_data);
	connection_timer.async_wait(check_connections);
	report_timer.async_wait(generate_report);
	
	signals.async_wait(signal_received);
	
	broadcast_timer.expires_from_now(boost::posix_time::milliseconds(tx_interval));
	broadcast_timer.async_wait(broadcast_data);
	
	io_service.run();
}

	void async_rename(const boost::filesystem::path & source,
		const boost::filesystem::path & destination,
		Callback callback)
	{
		worker_service_.post(boost::bind(
			&filesystem_service::rename<boost::_bi::protected_bind_t<Callback> >,
			this,
			boost::make_shared<boost::asio::io_service::work>(boost::ref(io_service_)),
			source,
			destination,
			boost::protect(callback)
		));
	}

static inline void avloop_run(boost::asio::io_service& io_service)
{
	using namespace ::detail;

	if (!boost::asio::has_service<IdleService>(io_service))
		boost::asio::add_service(io_service, new IdleService(io_service));

	while (!io_service.stopped())
	{
		if(!boost::asio::use_service<IdleService>(io_service).has_idle())
		{
			if (!io_service.run_one())
				break;
		}
		else
		{
			while (io_service.poll());
			// 执行 idle handler!
			boost::asio::use_service<IdleService>(io_service).poll_one();
		}
	}
}

void Read_arduino()
{
    //program timer for write operations
    tim.expires_from_now(boost::posix_time::seconds(5));
    tim.async_wait(timer_handler);
    //program chain of read operations
    async_read_until(sp,read_buf,'\n',read_handler);
    io.run();
}

void stopLogThread()
{
    if (LogThreadWork.get())
    {
        LogThreadWork.reset();
        LogIoService.stop();
        LogThread.try_join_for(boost::chrono::milliseconds(100));
    }
}

int main(int argc, char* argv[]) {


	t.async_wait(&print);

	io.run();

	return EXIT_SUCCESS;
}