MORDOR_UNITTEST(SSLStream, basic)
{
    WorkerPool pool;
    std::pair<Stream::ptr, Stream::ptr> pipes = pipeStream();

    SSLStream::ptr sslserver(new SSLStream(pipes.first, false));
    SSLStream::ptr sslclient(new SSLStream(pipes.second, true));

    pool.schedule(boost::bind(&accept, sslserver));
    sslclient->connect();
    pool.dispatch();

    Stream::ptr server = sslserver, client = sslclient;

    char buf[6];
    buf[5] = '\0';
    client->write("hello");
    client->flush(false);
    MORDOR_TEST_ASSERT_EQUAL(server->read(buf, 5), 5u);
    MORDOR_TEST_ASSERT_EQUAL((const char *)buf, "hello");
    server->write("world");
    server->flush(false);
    MORDOR_TEST_ASSERT_EQUAL(client->read(buf, 5), 5u);
    MORDOR_TEST_ASSERT_EQUAL((const char *)buf, "world");
}

int main()
{
    WorkerPool workers;

    for (int i = 0; i != 100; ++i)
    {
        workers.schedule([=]{ usleep(10); }, 0);
    }

    for (int i = 0; i != 50; ++i)
    {
        workers.schedule([=]{ usleep(10); }, 1);
    }

    for (int i = 0; i != 40; ++i)
    {
        workers.schedule([=]{ usleep(10); }, 2);
    }

    for (int i = 0; i != 10; ++i)
    {
        workers.schedule([=]{ usleep(10); }, 3);
    }
    
    sleep(1);
    quit = true;
}

// Just calling stop should still clear all pending work
MORDOR_UNITTEST(Scheduler, hijackStopOnScheduled)
{
    Fiber::ptr doNothingFiber(new Fiber(&doNothing));
    WorkerPool pool;
    MORDOR_TEST_ASSERT_EQUAL(Scheduler::getThis(), &pool);
    pool.schedule(doNothingFiber);
    MORDOR_TEST_ASSERT_EQUAL(doNothingFiber->state(), Fiber::INIT);
    pool.stop();
    MORDOR_TEST_ASSERT_EQUAL(doNothingFiber->state(), Fiber::TERM);
}

MORDOR_UNITTEST(PipeStream, closeOnBlockingReader)
{
    std::pair<Stream::ptr, Stream::ptr> pipe = pipeStream();
    WorkerPool pool;
    int sequence = 1;

    pool.schedule(Fiber::ptr(new Fiber(boost::bind(&closeOnBlockingReader,
                                       pipe.first, boost::ref(sequence)))));

    Buffer output;
    MORDOR_TEST_ASSERT_EQUAL(pipe.second->read(output, 10), 0u);
    MORDOR_TEST_ASSERT_EQUAL(++sequence, 4);
}

int main(int argc, char const *argv[])
{
	WorkerPool wp;
	std::cerr << "pool created\n";
	wp.waitallready();
	std::cerr << "pool ready\n";
	Scheduler s(argc == 3 ? 0: &wp,argv[1]);

	int q = 0;

	s << 1 << [&] {
		{lockio x; std::cerr << "i1A " << q << "\n"; }
		q = 1;
		usleep(10000);
		{lockio x; std::cerr << "i1B " << q << "\n"; }
	}	;

	s << 2 << [&] {
		{lockio x; std::cerr << "i2A " << q << "\n"; }
		q = 2;
		usleep(20000);
		{lockio x; std::cerr << "i2B " << q << "\n"; }
	}	;

	s << 3 <<  [&] {
		{lockio x; std::cerr << "i3A " << q << "\n"; }
		q = 3;
		usleep(30000);
		{lockio x; std::cerr << "i3B " << q << "\n"; }
	}	;

	// 
	s << 4 <<  [&] (int from, int to) {
		{lockio x; std::cerr << "i4A " << q <<  " range " << from << "-" << to << "\n"; }
		q = 4;
		usleep(30000);
		{lockio x; std::cerr << "i4B " << q << " range " << from << "-" << to <<  "\n"; }
	}	;


	std::cerr << "before run\n";
	s.run();
	std::cerr << "after run\n";
	return 0;
}

void run(int argc, char **argv){
	ready_list_t ready_list;
	ready_list_t ready_list_2;
	ready_list_t::iterator it;
	const Fdevents::events_t *events;
	Server serv(ssdb);

	Fdevents select;
	select.set(serv_link->fd(), FDEVENT_IN, 0, serv_link);
	select.set(serv.reader->fd(), FDEVENT_IN, 0, serv.reader);
	select.set(serv.writer->fd(), FDEVENT_IN, 0, serv.writer);
	
	int link_count = 0;
	while(!quit){
		bool write_pending = false;
		ready_list.clear();
		ready_list_2.clear();
		
		if(write_pending || !ready_list.empty()){
			// give links that are not in ready_list a chance
			events = select.wait(0);
		}else{
			events = select.wait(50);
		}
		if(events == NULL){
			log_fatal("events.wait error: %s", strerror(errno));
			break;
		}
		for(int i=0; i<(int)events->size(); i++){
			const Fdevent *fde = events->at(i);
			if(fde->data.ptr == serv_link){
				Link *link = serv_link->accept();
				if(link == NULL){
					log_error("accept fail!");
					continue;
				}
				link_count ++;
				log_info("new link from %s:%d, fd: %d, link_count: %d",
					link->remote_ip, link->remote_port, link->fd(), link_count);
				
				link->nodelay();
				link->noblock();
				link->create_time = millitime();
				link->active_time = link->create_time;
				select.set(link->fd(), FDEVENT_IN, 1, link);
			}else if(fde->data.ptr == serv.reader || fde->data.ptr == serv.writer){
				WorkerPool<Server::ProcWorker, ProcJob> *worker = (WorkerPool<Server::ProcWorker, ProcJob> *)fde->data.ptr;
				ProcJob job;
				if(worker->pop(&job) == 0){
					log_fatal("reading result from workers error!");
					exit(0);
				}
				if(proc_result(job, select, ready_list_2) == PROC_ERROR){
					link_count --;
				}
			}else{
				Link *link = (Link *)fde->data.ptr;
				// 不能同时监听读和写事件, 只能监听其中一个
				if(fde->events & FDEVENT_ERR){
					log_info("fd: %d error, delete link", link->fd());
					link_count --;
					select.del(link->fd());
					delete link;
				}else if(fde->events & FDEVENT_IN){
					int len = link->read();
					//log_trace("fd: %d read: %d", link->fd(), len);
					if(len <= 0){
						log_info("fd: %d, read: %d, delete link", link->fd(), len);
						link_count --;
						select.del(link->fd());
						delete link;
					}else{
						ready_list.push_back(link);
					}
				}else if(fde->events & FDEVENT_OUT){
					int len = link->write();
					//log_trace("fd: %d write: %d", link->fd(), len);
					if(len <= 0){
						log_info("fd: %d, write: %d, delete link", link->fd(), len);
						link_count --;
						select.del(link->fd());
						delete link;
					}else if(link->output->empty()){
						//log_trace("delete %d from select.out", link->fd());
						select.clr(link->fd(), FDEVENT_OUT);
						if(!link->input->empty()){
							ready_list.push_back(link);
						}else{
							//log_trace("add %d to select.in", link->fd());
							select.set(link->fd(), FDEVENT_IN, 1, link);
						}
					}else{
						write_pending = true;
					}
				}
			}
		}

		for(it = ready_list.begin(); it != ready_list.end(); it ++){
			Link *link = *it;
//.........这里部分代码省略.........