<|start_filename|>unit_test/test_server.cpp<|end_filename|>
// ====================================================================================
// Copyright (c) 2012, <EMAIL> All rights reserved.
// File         : test_server.cpp
// Author       : <EMAIL>
// Last Change  : 11/17/2012 06:11 PM
// Description  : 
// ====================================================================================


#include <iterator>
#include <algorithm>
#include <string>
#include <iostream>
#include <boost/shared_ptr.hpp>

#include "../orchid/all.hpp"


using std::string;
using std::cout;
using std::cerr;
using std::endl;


const static std::size_t STACK_SIZE = 16*1024;



typedef boost::shared_ptr<orchid::socket> socket_ptr;

void handle_io(orchid::coroutine_handle co,socket_ptr sock) {
    orchid::buffered_reader<orchid::socket> reader(*sock,co,16);
    orchid::buffered_writer<orchid::socket> writer(*sock,co,16);

    try {
        std::string line;
        std::size_t n = 0;

        for(;;) {
            n = reader.read_until(line,'\n');
            ORCHID_DEBUG("id %lu recv: %s",co->id(),line.c_str());
            writer.write(line.c_str(),line.size());
            writer.flush();
        }

    } catch (const orchid::io_error& e) {
        if (e.code() == boost::asio::error::eof) {
            ORCHID_DEBUG("id %lu msg:%s",co->id(),"socket closed by remote side!");
        } else {
            ORCHID_ERROR("id %lu msg:%s",co->id(),e.what());
        }
    }

}

void handle_accept(orchid::coroutine_handle co) {
    try {
        orchid::acceptor acceptor(co -> get_io_service());
        acceptor.bind_and_listen("5678",false);
        for(;;) {
            socket_ptr sock(new orchid::socket(co -> get_io_service()));
            acceptor.accept(*sock,co);
            co -> get_scheduler().spawn(boost::bind(handle_io,_1,sock));
        }
    }
    catch(orchid::io_error& e) {
        ORCHID_ERROR("id %lu msg:%s",co->id(),e.what());
    }
}

int main() {

    orchid::scheduler sche;
    ORCHID_DEBUG("id:%lu",sche.id());
    sche.spawn(handle_accept,orchid::coroutine::minimum_stack_size());
    sche.run();
}

<|start_filename|>orchid/coroutine/scheduler_group.hpp<|end_filename|>
// ====================================================================================
// Copyright (c) 2012, <EMAIL> All rights reserved.
// File         : scheduler_group_basic.hpp
// Author       : <EMAIL>
// Last Change  : 11/21/2012 08:47 PM
// Description  : 
// ====================================================================================

#ifndef __ORCHID_SCHEDULER_GROUP_H__
#define __ORCHID_SCHEDULER_GROUP_H__

#include <boost/thread.hpp>
#include <boost/bind.hpp>
#include <boost/scoped_array.hpp>
#include <boost/assert.hpp>
#include <boost/atomic.hpp>
// #include <boost/random.hpp>

namespace orchid { namespace detail {
template <typename Scheduler>
class scheduler_group_basic {
public:
    typedef Scheduler scheduler_type;
    typedef typename scheduler_type::io_service_type io_service_type;

public:
    scheduler_group_basic(std::size_t size)
        :size_(size),sches_(new scheduler_type[size]) {

    }
    ~scheduler_group_basic() {
    }
public:
    void run() {
        for(std::size_t i = 0;i < size_;++i) {
            tg_.create_thread(boost::bind(&scheduler_type::run,&sches_[i]));
        }
        tg_.join_all();
    }

    void stop() {
        for(std::size_t i=0;i<size_;++i) {
            sches_[i].stop();
        }
    }

    std::size_t size() {
        return size_;
    }

    //
    scheduler_type& operator[](std::size_t i) {
        return sches_[i];
    }

    const scheduler_type& operator[](std::size_t i) const {
        return sches_[i];
    }


private:

    std::size_t size_;
    boost::thread_group tg_;
    boost::scoped_array<scheduler_type> sches_;

};


}}


#endif


<|start_filename|>orchid/asio/io_funcs.hpp<|end_filename|>
// ====================================================================================
// Copyright (c) 2012, <EMAIL> All rights reserved.
// File         : io_funcs.hpp
// Author       : <EMAIL>
// Last Change  : 11/16/2012 05:58 PM
// Description  : 
// ====================================================================================
#ifndef __ORCHID_IO_FUNCS_H__
#define __ORCHID_IO_FUNCS_H__

#include <iostream>
#include <boost/asio.hpp>
#include <boost/utility.hpp>
#include <boost/system/error_code.hpp>
#include <boost/system/system_error.hpp>
#include <boost/assert.hpp>
#include <boost/regex.hpp>

#include "handlers.hpp"
#include "throw_error.hpp"
#include "../utility/debug.hpp"


namespace orchid { 

    template<typename AsyncReadStream,
            typename MutableBufferSequence,
            typename CO>
    std::size_t read(AsyncReadStream& s,
                    const MutableBufferSequence & buffers,
                    CO co,
                    boost::system::error_code& e) {

        BOOST_ASSERT(co != NULL);
        std::size_t bytes_transferred;
        detail::io_handler<CO> handler(co,e,bytes_transferred);
        boost::asio::async_read(s,buffers,handler);
        //////////////////////////////////
        co->yield();
        //////////////////////////////////
        if(e) {
            ORCHID_DEBUG("io_funcs read error: %s",e.message().c_str());
        }
        return bytes_transferred; 

    }

    template<typename AsyncReadStream,
            typename MutableBufferSequence,
            typename CO>
    std::size_t read(AsyncReadStream& s,
                    const MutableBufferSequence & buffers,
                    CO co) {

        boost::system::error_code e;
        std::size_t bytes_transferred = orchid::read(s,buffers,co,e);
        if (e) {
            detail::throw_error(e,"io_funcs read");
        }

        return bytes_transferred;
    }



    template<typename AsyncReadStream,
            typename MutableBufferSequence,
            typename CompletionCondition,
            typename CO>
    std::size_t read(AsyncReadStream& s,
                    const MutableBufferSequence & buffers,
                    CompletionCondition completion_condition,
                    CO co,
                    boost::system::error_code& e) {

        BOOST_ASSERT(co != NULL);
        std::size_t bytes_transferred;
        detail::io_handler<CO> handler(co,e,bytes_transferred);
        boost::asio::async_read(s,buffers,completion_condition,handler);
        //////////////////////////////////
        co->yield();
        //////////////////////////////////
        if(e) {
            ORCHID_DEBUG("io_funcs read error: %s",e.message().c_str());
        }
        return bytes_transferred; 

    }


    template<typename AsyncReadStream,
            typename MutableBufferSequence,
            typename CompletionCondition,
            typename CO>
    std::size_t read(AsyncReadStream& s,
                    const MutableBufferSequence & buffers,
                    CompletionCondition completion_condition,
                    CO co) {

        BOOST_ASSERT(co != NULL);
        boost::system::error_code e;
        std::size_t bytes_transferred = orchid::read(s,buffers,completion_condition,co,e);
        if (e) {
            detail::throw_error(e,"io_funcs read");
        }

        return bytes_transferred;

    }

    template<typename AsyncReadStream,typename Allocator,typename CO>
    std::size_t read(AsyncReadStream& s,
                    boost::asio::basic_streambuf< Allocator > & b,
                    CO co,
                    boost::system::error_code& e) {

        BOOST_ASSERT(co != NULL);
        std::size_t bytes_transferred;
        detail::io_handler<CO> handler(co,e,bytes_transferred);
        boost::asio::async_read(s,b,handler);
        //////////////////////////////////
        co->yield();
        //////////////////////////////////
        if(e) {
            ORCHID_DEBUG("io_funcs read error: %s",e.message().c_str());
        }
        return bytes_transferred; 

    }

    template<typename AsyncReadStream,typename Allocator,typename CO>
    std::size_t read(AsyncReadStream& s,boost::asio::basic_streambuf< Allocator > & b,CO co) {

        BOOST_ASSERT(co != NULL);
        boost::system::error_code e;
        std::size_t bytes_transferred = orchid::read(s,b,co,e);
        if (e) {
            detail::throw_error(e,"io_funcs read");
        }

        return bytes_transferred;

    }



    template<typename AsyncReadStream,typename Allocator,typename CompletionCondition,typename CO>
    std::size_t read(AsyncReadStream& s,
                    boost::asio::basic_streambuf< Allocator > & b,
                    CompletionCondition completion_condition,
                    CO co,
                    boost::system::error_code& e) {

        BOOST_ASSERT(co != NULL);
        std::size_t bytes_transferred;
        detail::io_handler<CO> handler(co,e,bytes_transferred);
        boost::asio::async_read(s,b,completion_condition,handler);
        //////////////////////////////////
        co->yield();
        //////////////////////////////////
        if(e) {
            ORCHID_DEBUG("io_funcs read error: %s",e.message().c_str());
        }
        return bytes_transferred; 

    }

    template<typename AsyncReadStream,typename Allocator,typename CompletionCondition,typename CO>
    std::size_t read(AsyncReadStream& s,
                    boost::asio::basic_streambuf< Allocator > & b,
                    CompletionCondition completion_condition,
                    CO co) {

        BOOST_ASSERT(co != NULL);
        boost::system::error_code e;
        std::size_t bytes_transferred = orchid::read(s,b,completion_condition,co,e);
        if (e) {
            detail::throw_error(e,"io_funcs read");
        }

        return bytes_transferred;

    }



    template<typename AsyncReadStream,
            typename Allocator,
            typename CO>
    std::size_t read_until(AsyncReadStream & s,
                boost::asio::basic_streambuf< Allocator > & b,
                char delim,
                CO co,
                boost::system::error_code& e) {

        BOOST_ASSERT(co != NULL);
        std::size_t bytes_transferred;
        detail::io_handler<CO> handler(co,e,bytes_transferred);
        boost::asio::async_read_until(s,b,delim,handler);
        //////////////////////////////////
        co->yield();
        //////////////////////////////////
        if(e) {
            ORCHID_DEBUG("io_funcs read_until error: %s",e.message().c_str());
        }
        return bytes_transferred; 

    }

    template<typename AsyncReadStream,
            typename Allocator,
            typename CO>
    std::size_t read_until(AsyncReadStream & s,
                boost::asio::basic_streambuf< Allocator > & b,
                char delim,
                CO co) {

        BOOST_ASSERT(co != NULL);
        boost::system::error_code e;
        std::size_t bytes_transferred = orchid::read_until(s,b,delim,co,e);
        if (e) {
            detail::throw_error(e,"io_funcs read_until");
        }

        return bytes_transferred;

    }


    template<typename AsyncReadStream,
            typename Allocator,
            typename CO>
    std::size_t read_until(AsyncReadStream & s,
                boost::asio::basic_streambuf< Allocator > & b,
                const std::string& delim,
                CO co,
                boost::system::error_code& e) {

        BOOST_ASSERT(co != NULL);
        std::size_t bytes_transferred;
        detail::io_handler<CO> handler(co,e,bytes_transferred);
        boost::asio::async_read_until(s,b,delim,handler);
        //////////////////////////////////
        co->yield();
        //////////////////////////////////
        if(e) {
            ORCHID_DEBUG("io_funcs read_until error: %s",e.message().c_str());
        }
        return bytes_transferred; 

    }

    template<typename AsyncReadStream,
            typename Allocator,
            typename CO>
    std::size_t read_until(AsyncReadStream & s,
                boost::asio::basic_streambuf< Allocator > & b,
                const std::string& delim,
                CO co) {

        BOOST_ASSERT(co != NULL);
        boost::system::error_code e;
        std::size_t bytes_transferred = orchid::read_until(s,b,delim,co,e);
        if (e) {
            detail::throw_error(e,"io_funcs read_until");
        }

        return bytes_transferred;

    }


    template<typename AsyncReadStream,
            typename Allocator,
            typename CO>
    std::size_t read_until(AsyncReadStream & s,
                boost::asio::basic_streambuf< Allocator > & b,
                const boost::regex& delim,
                CO co,
                boost::system::error_code& e) {

        BOOST_ASSERT(co != NULL);
        std::size_t bytes_transferred;
        detail::io_handler<CO> handler(co,e,bytes_transferred);
        boost::asio::async_read_until(s,b,delim,handler);
        //////////////////////////////////
        co->yield();
        //////////////////////////////////
        if(e) {
            ORCHID_DEBUG("io_funcs read_until error: %s",e.message().c_str());
        }
        return bytes_transferred; 

    }

    template<typename AsyncReadStream,
            typename Allocator,
            typename CO>
    std::size_t read_until(AsyncReadStream & s,
                boost::asio::basic_streambuf< Allocator > & b,
                const boost::regex& delim,
                CO co) {

        BOOST_ASSERT(co != NULL);
        boost::system::error_code e;
        std::size_t bytes_transferred = orchid::read_until(s,b,delim,co,e);
        if (e) {
            detail::throw_error(e,"io_funcs read_until");
        }

        return bytes_transferred;

    }

    //////////////////////////////////////////////////////////
    template<typename AsyncWriteStream,
            typename ConstBufferSequence,
            typename CO>
    std::size_t write(AsyncWriteStream & s,
                const ConstBufferSequence & buffers,
                CO co,
                boost::system::error_code& e) {

        BOOST_ASSERT(co != NULL);
        std::size_t bytes_transferred;
        detail::io_handler<CO> handler(co,e,bytes_transferred);
        boost::asio::async_write(s,buffers,handler);
        //////////////////////////////////
        co->yield();
        //////////////////////////////////
        if(e) {
            ORCHID_DEBUG("io_funcs write error: %s",e.message().c_str());
        }
        return bytes_transferred; 

    }


    template<typename AsyncWriteStream,
            typename ConstBufferSequence,
            typename CO>
    std::size_t write(AsyncWriteStream & s,
                const ConstBufferSequence & buffers,
                CO co) {

        BOOST_ASSERT(co != NULL);
        boost::system::error_code e;
        std::size_t bytes_transferred = orchid::write(s,buffers,co,e);
        if (e) {
            detail::throw_error(e,"io_funcs write");
        }

        return bytes_transferred;

    }

    template<typename AsyncWriteStream,
            typename ConstBufferSequence,
            typename CompletionCondition,
            typename CO>
    std::size_t write(AsyncWriteStream & s,
                const ConstBufferSequence & buffers,
                CompletionCondition completion_condition,
                CO co,
                boost::system::error_code& e) {

        BOOST_ASSERT(co != NULL);
        std::size_t bytes_transferred;
        detail::io_handler<CO> handler(co,e,bytes_transferred);
        boost::asio::async_write(s,buffers,completion_condition,handler);
        //////////////////////////////////
        co->yield();
        //////////////////////////////////
        if(e) {
            ORCHID_DEBUG("io_funcs write error: %s",e.message().c_str());
        }
        return bytes_transferred; 

    }


    template<typename AsyncWriteStream,
            typename ConstBufferSequence,
            typename CompletionCondition,
            typename CO>
    std::size_t write(AsyncWriteStream & s,
                const ConstBufferSequence & buffers,
                CompletionCondition completion_condition,
                CO co) {

        BOOST_ASSERT(co != NULL);
        boost::system::error_code e;
        std::size_t bytes_transferred = orchid::write(s,buffers,completion_condition,co,e);
        if (e) {
            detail::throw_error(e,"io_funcs write");
        }

        return bytes_transferred;

    }


    template<typename AsyncWriteStream,
            typename Allocator,
            typename CO>
    std::size_t write(AsyncWriteStream & s,
                boost::asio::basic_streambuf< Allocator > & b,
                CO co,
                boost::system::error_code& e) {

        BOOST_ASSERT(co != NULL);
        std::size_t bytes_transferred;
        detail::io_handler<CO> handler(co,e,bytes_transferred);
        boost::asio::async_write(s,b,handler);
        //////////////////////////////////
        co->yield();
        //////////////////////////////////
        if(e) {
            ORCHID_DEBUG("io_funcs write error: %s",e.message().c_str());
        }
        return bytes_transferred; 

    }


    template<typename AsyncWriteStream,
            typename Allocator,
            typename CO>
    std::size_t write(AsyncWriteStream & s,
                boost::asio::basic_streambuf< Allocator > & b,
                CO co) {

        BOOST_ASSERT(co != NULL);
        boost::system::error_code e;
        std::size_t bytes_transferred = orchid::write(s,b,co,e);
        if (e) {
            detail::throw_error(e,"io_funcs write");
        }

        return bytes_transferred;

    }


    template<typename AsyncWriteStream,
            typename Allocator,
            typename CompletionCondition,
            typename CO>
    std::size_t write(AsyncWriteStream & s,
                boost::asio::basic_streambuf< Allocator > & b,
                CompletionCondition completion_condition,
                CO co,
                boost::system::error_code& e) {

        BOOST_ASSERT(co != NULL);
        std::size_t bytes_transferred;
        detail::io_handler<CO> handler(co,e,bytes_transferred);
        boost::asio::async_write(s,b,completion_condition,handler);
        //////////////////////////////////
        co->yield();
        //////////////////////////////////
        if(e) {
            ORCHID_DEBUG("io_funcs write error: %s",e.message().c_str());
        }
        return bytes_transferred; 

    }


    template<typename AsyncWriteStream,
            typename Allocator,
            typename CompletionCondition,
            typename CO>
    std::size_t write(AsyncWriteStream & s,
                boost::asio::basic_streambuf< Allocator > & b,
                CompletionCondition completion_condition,
                CO co) {

        BOOST_ASSERT(co != NULL);
        boost::system::error_code e;
        std::size_t bytes_transferred = orchid::write(s,b,completion_condition,co,e);
        if (e) {
            detail::throw_error(e,"io_funcs write");
        }

        return bytes_transferred;

    }


}

#endif




<|start_filename|>orchid/coroutine/scheduler.hpp<|end_filename|>
// ====================================================================================
// Copyright (c) 2012, <EMAIL> All rights reserved.
// File         : scheduler.hpp
// Author       : <EMAIL>
// Last Change  : 11/08/2012 12:35 AM
// Description  : 
// ====================================================================================
#ifndef __ORCHID_SCHEDULER_H__
#define __ORCHID_SCHEDULER_H__

#include <set>
#include <algorithm>
#include <iostream>

#include <boost/assert.hpp>
#include <boost/function.hpp>
#include <boost/context/all.hpp>
#include <boost/utility.hpp>
#include <boost/bind.hpp>
#include <boost/function.hpp>
#include <boost/atomic.hpp>

#include "stack_allocator.hpp"
#include "../utility/debug.hpp"

namespace orchid { namespace detail {

template <template <class,class> class Coroutine,typename IOservice,typename Alloc = stack_allocator>
class scheduler_basic:boost::noncopyable {
public:
    typedef scheduler_basic<Coroutine,IOservice,Alloc> self_type;
    typedef IOservice io_service_type;
    typedef Alloc allocator_type;
    typedef Coroutine<self_type,allocator_type> coroutine_type;
    typedef boost::shared_ptr<coroutine_type> coroutine_pointer;
    typedef boost::context::fcontext_t context_type;
public:
    static boost::atomic<unsigned long> scheduler_id_gen_;

public:
	scheduler_basic() :id_(scheduler_id_gen_.fetch_add(1)), coroutine_id_gen_(0) {
        ORCHID_DEBUG("sche %lu: scheduler_basic()",id());
    }
    ~scheduler_basic() {
        ORCHID_DEBUG("sche %lu: ~scheduler_basic()",id());
        typename std::set<coroutine_pointer>::iterator it;
        for(it = all_.begin();it != all_.end();++it) {
            (*it) -> stop();
            ORCHID_DEBUG("sche %lu: clear coroutine %lu",id(),(*it)->id());
            boost::context::jump_fcontext(&ctx_,(*it)->ctx(),(intptr_t)((*it).get()));
            BOOST_ASSERT((*it) -> is_dead());
        }
        all_.clear();
    }

public:

    void run() {
        ORCHID_DEBUG("sche %lu run",id_);
        io_service_.run();
    }

    void stop() {
        io_service_.stop();
        // typename std::set<coroutine_pointer>::iterator it;
        // for(it = all_.begin();it != all_.end();++it) {
        //     (*it) -> stop();
        // }
    }

    template <typename F>
    void spawn(const F& f,std::size_t stack_size = coroutine_type::default_stack_size()) {
        coroutine_pointer co(new coroutine_type(*this,coroutine_id_gen_.fetch_add(1),f,stack_size));
        io_service_.post(boost::bind(&self_type::do_spawn,this,co));
        ORCHID_DEBUG("sche %lu spawn done",id_);
    }

    void resume(coroutine_pointer co) {
        BOOST_ASSERT(co != NULL);
        do_schedule(co);
    }

    template <typename F>
    void post(const F& f) {
        io_service_.post(f);
    }

    io_service_type& get_io_service() {
        return io_service_;
    }

    const io_service_type& get_io_service() const {
        return io_service_;
    }
    /////////////////////////////////////////////////

    context_type& ctx() {
        return ctx_;
    }

    const context_type& ctx() const {
        return ctx_;
    }

    unsigned long id() const {
        return id_;
    }

private:

    void do_schedule(coroutine_pointer co) {
        //jump to coroutine
        ORCHID_DEBUG("sche:%lu,coroutine:%lu,resume",id(),co->id());
        boost::context::jump_fcontext(&ctx_,co->ctx(),(intptr_t)(co.get()));
        if(co -> is_dead()) {
            all_.erase(co);
        }
    }

    void do_spawn(coroutine_pointer co) {
        ORCHID_DEBUG("sche %lu do spawn",id_);
        all_.insert(co);
        do_schedule(co);
    }

private:
    const unsigned long id_;
    boost::atomic<unsigned long> coroutine_id_gen_;
    context_type ctx_;
    std::set<coroutine_pointer> all_;
    io_service_type io_service_;
};



template <template <class,class> class Coroutine,typename IOservice,typename Alloc>
boost::atomic<unsigned long> scheduler_basic<Coroutine,IOservice,Alloc>::scheduler_id_gen_(0);


} }

#endif


<|start_filename|>orchid/all.hpp<|end_filename|>
// ====================================================================================
// Copyright (c) 2012, <EMAIL> All rights reserved.
// File         : all.hpp
// Author       : <EMAIL>
// Last Change  : 11/16/2012 10:12 PM
// Description  : 
// ====================================================================================

#ifndef __ORCHID_ALL_H__
#define __ORCHID_ALL_H__

#define USE_BOOST_ASIO

#include <memory>

#include "coroutine/scheduler.hpp"
#include "coroutine/coroutine.hpp"
#include "coroutine/chan.hpp"
#include "coroutine/stack_allocator.hpp"
#include "coroutine/scheduler_group.hpp"
#include "coroutine/chan_map.hpp"

#include "asio/acceptor.hpp"
#include "asio/socket.hpp"
#include "asio/timer.hpp"
#include "asio/io_service.hpp"
#include "asio/descriptor.hpp"
#include "asio/signal.hpp"
#include "asio/io_interface.hpp"
#include "asio/io_funcs.hpp"
#include "asio/throw_error.hpp"
#include "asio/worker.hpp"

#include "utility/debug.hpp"

// #if defined(__linux__)
// #endif

namespace orchid {

typedef detail::scheduler_basic<detail::coroutine_basic,detail::io_service> scheduler;
typedef detail::scheduler_group_basic<scheduler> scheduler_group;
typedef detail::coroutine_basic<scheduler> coroutine;
typedef coroutine::coroutine_handle coroutine_handle;
typedef detail::socket_basic socket;
typedef detail::acceptor_basic acceptor;
typedef detail::timer_basic timer;
typedef detail::descriptor_basic descriptor;
typedef detail::signal_basic signal;
typedef detail::io_error io_error;
typedef detail::worker_pool_basic worker_pool;


template <typename Input>
class reader:public detail::reader_basic<coroutine_handle,Input> {
public:
    reader(Input& input,coroutine_handle co)
        :detail::reader_basic<coroutine_handle,Input>(input,co) {

        }

    ~reader() {

    }
};


template <typename Output>
class writer:public detail::writer_basic<coroutine_handle,Output> {
public:
    writer(Output& output,coroutine_handle co)
        :detail::writer_basic<coroutine_handle,Output>(output,co) {

        }

    ~writer() {

    }

};

template <typename Input,typename Alloc=std::allocator<char> >
class buffered_reader:public detail::buffered_reader_basic<coroutine_handle,Input,Alloc> {
public:
    buffered_reader(Input& input,coroutine_handle co,std::size_t size=detail::buffered_reader_basic<coroutine_handle,Input,Alloc>::default_size)
        :detail::buffered_reader_basic<coroutine_handle,Input,Alloc>(input,co) {

    }

    ~buffered_reader() {

    }

};

template <typename Output,typename Alloc=std::allocator<char> >
class buffered_writer:public detail::buffered_writer_basic<coroutine_handle,Output,Alloc> {
public:
    buffered_writer(Output& output,coroutine_handle co,std::size_t size=detail::buffered_reader_basic<coroutine_handle,Output,Alloc>::default_size)
        :detail::buffered_writer_basic<coroutine_handle,Output,Alloc>(output,co) {

    }

    ~buffered_writer() {

    }

};


template <typename T>
class chan:public detail::chan_basic<coroutine_handle,T> {
public:
    typedef chan<T> self_type;
public:
    chan(std::size_t cap)
        :detail::chan_basic<coroutine_handle,T>(cap) {

    }
    ~chan() {

    }
};


template <typename K, typename T>
class chan_map :public detail::chan_map_basic<coroutine_handle, K, T> {
public:
	typedef chan_map<K, T> self_type;
public:
	chan_map(std::size_t chan_cap)
		:detail::chan_map_basic<coroutine_handle, K, T>(chan_cap) {

		}
	~chan_map() {

	}
};

}

#endif


<|start_filename|>orchid/coroutine/chan_map.hpp<|end_filename|>
// ====================================================================================
// Copyright (c) 2012, <EMAIL> All rights reserved.
// File         : chan.hpp
// Author       : <EMAIL>
// Last Change  : 11/18/2012 01:35 AM
// Description  : 
// ====================================================================================

#ifndef __ORCHID_CHAN_MAP_H__
#define __ORCHID_CHAN_MAP_H__

#include <iostream>
#include <memory>
#include <map>

#include <boost/utility.hpp>
#include <boost/shared_ptr.hpp>

#include "chan.hpp"
#include "spinlock.hpp"


namespace orchid {
	namespace detail {
		
		template <typename CO, typename K, typename T>
		class chan_map_basic :boost::noncopyable
		{
		public:
			typedef chan_map_basic<CO, K, T> self_type;
			typedef chan_basic<CO, T> chan_type;
			typedef K key_type;
			typedef T value_type;
			typedef CO coroutine_pointer;
		public:
			chan_map_basic(std::size_t chan_cap)
				:chan_cap_(chan_cap) {
			}
			~chan_map_basic() {
			}
		public:
			bool open(const K& k)
			{
				locker_.lock();
				std::pair<typename std::map<key_type, boost::shared_ptr<chan_type> >::iterator, bool> ret = map_.insert(std::make_pair(k, boost::shared_ptr<chan_type>()));
				if (ret.second)
					ret.first->second.reset(new chan_type(1));
				locker_.unlock();
				return ret.second;
			}
			void close(const K& k) {
				locker_.lock();
				typename std::map<key_type, boost::shared_ptr<chan_type> >::iterator it = map_.find(k);
				if (it != map_.end())
				{
					it->second->close();
					map_.erase(it);
				}
				locker_.unlock();
			}
			bool send(const K& k, const T& t, coroutine_pointer co) {
				bool ret = false;
				locker_.lock();
				typename std::map<key_type, boost::shared_ptr<chan_type> >::iterator it = map_.find(k);
				if (it != map_.end())
				{
					boost::shared_ptr<chan_type> chanptr(it->second);
					locker_.unlock();
					ret = chanptr->send(t, co);
				}
				else
				{
					locker_.unlock();
				}
				return ret;
			}
			bool recv(const K& k, T& t, coroutine_pointer co) {
				bool ret = false;
				locker_.lock();
				typename std::map<key_type, boost::shared_ptr<chan_type> >::iterator it = map_.find(k);
				if (it != map_.end())
				{
					boost::shared_ptr<chan_type> chanptr(it->second);
					locker_.unlock();
					ret = chanptr->recv(t, co);
				}
				else
				{
					locker_.unlock();
				}
				return ret;
			}
		private:
			spinlock locker_;
			std::map<key_type, boost::shared_ptr<chan_type> > map_;
			std::size_t chan_cap_;
		};



	}
}

#endif//__ORCHID_CHAN_MAP_H__

<|start_filename|>unit_test/hello_world.cpp<|end_filename|>
// ====================================================================================
// Copyright (c) 2012, <EMAIL> All rights reserved.
// File         : hello_world.cpp
// Author       : <EMAIL>
// Last Change  : 12/18/2012 11:18 PM
// Description  : 
// ====================================================================================


#include <unistd.h>
#include <stdio.h>
#include <iostream>
#include <string>

#include "../orchid/all.hpp"


void f(orchid::coroutine_handle co) {
    std::cout<<"f:hello world"<<std::endl;
}

void f1(orchid::coroutine_handle co,const char* str) {
    std::cout<<str<<std::endl;
}

struct printer {
    printer(const std::string& s):str(s) {

    }
    void operator()(orchid::coroutine_handle co) {
        std::cout<<str<<std::endl;
    }
    std::string str;
};

void stop(orchid::coroutine_handle co) {
    co -> get_scheduler().stop();
}

int main(int argc,char* argv[]) {
    orchid::scheduler sche;
    printer f2("f2:hello world");
    sche.spawn(f,orchid::coroutine::minimum_stack_size());
    sche.spawn(boost::bind(f1,_1,"f1:hello world"),orchid::coroutine::minimum_stack_size());
    sche.spawn(f2,orchid::coroutine::minimum_stack_size());
    sche.spawn(stop);
    sche.run();
    std::cout<<"done!"<<std::endl;
}

<|start_filename|>unit_test/test_chan.cpp<|end_filename|>
// ====================================================================================
// Copyright (c) 2012, <EMAIL> All rights reserved.
// File         : test_chan.cpp
// Author       : <EMAIL>
// Last Change  : 11/19/2012 12:35 AM
// Description  : 
// ====================================================================================

#include <unistd.h>
#include <string>
#include <stdio.h>
#include <iostream>
#include <boost/lexical_cast.hpp>
#include <boost/thread.hpp>
#include <boost/bind.hpp>

#include "../orchid/all.hpp"


using std::string;
using std::cout;
using std::endl;



const static std::size_t STACK_SIZE = 64*1024;



void sender(orchid::coroutine_handle co,int id,orchid::chan<int>& ch) {
    orchid::descriptor stdout(co -> get_io_service(),::dup(STDOUT_FILENO));
    orchid::writer<orchid::descriptor> console(stdout,co);
    char buf[128] = {0};

    for (;;) {
        ch.send(id,co);
        int n = sprintf(buf,"sender %d send:%d\r\n",id,id);
        console.write(buf,n);
    }
}

void receiver(orchid::coroutine_handle co,int id,orchid::chan<int>& ch) {
    orchid::descriptor stdout(co -> get_io_service(),::dup(STDOUT_FILENO));
    orchid::writer<orchid::descriptor> console(stdout,co);
    int sender = 0;
    char buf[128] = {0};

    for (;;) {
        ch.recv(sender,co);
        int n = sprintf(buf,"receiver %d receive:%d\r\n",id,sender);
        console.write(buf,n);
    }
}

void test_one_scheduler() {
    orchid::scheduler sche;
    orchid::chan<int> ch(100);
    for (int i=0;i<100;++i) {
        sche.spawn(boost::bind(sender,_1,i,boost::ref(ch)));
    }
    for (int i=0;i<100;++i) {
        sche.spawn(boost::bind(receiver,_1,i,boost::ref(ch)));
    }
    
    sche.run();
}

void test_scheduler_group() {
    int N = 2;
    orchid::scheduler_group group(N);
    orchid::chan<int> ch(10);
    for (int i=0;i<100;++i) {
        group[i%N].spawn(boost::bind(sender,_1,i,boost::ref(ch)));
    }
    for (int i=0;i<100;++i) {
        group[i%N].spawn(boost::bind(receiver,_1,i,boost::ref(ch)));
    }
    group.run();
    // boost::thread t(boost::bind(&orchid::scheduler_group::run,&group));
    // t.join()
}

int main(int argc,const char* argv[]) {
    // test_one_scheduler();
    test_scheduler_group();

}

 




<|start_filename|>orchid/asio/io_service.hpp<|end_filename|>
// ====================================================================================
// Copyright (c) 2012, <EMAIL> All rights reserved.
// File         : io_service.hpp
// Author       : <EMAIL>
// Last Change  : 11/21/2012 11:57 AM
// Description  : 
// ====================================================================================

#ifndef __ORCHID_IO_SERVICE_H__
#define __ORCHID_IO_SERVICE_H__

#include <boost/asio.hpp>
#include <boost/bind.hpp>
#include <boost/assert.hpp>
#include <boost/utility.hpp>

namespace orchid { namespace detail {


class io_service:public boost::asio::io_service {
    boost::asio::io_service::work work_;
public:
    io_service():boost::asio::io_service(),work_(*this) {

    }
    ~io_service() {
        
    }

    
};
    // typedef boost::asio::io_service io_service;

}}



#endif

<|start_filename|>orchid/asio/io_interface.hpp<|end_filename|>
// ====================================================================================
// Copyright (c) 2012, <EMAIL> All rights reserved.
// File         : io_interface.hpp
// Author       : <EMAIL>
// Last Change  : 11/17/2012 07:18 PM
// Description  : 
// ====================================================================================

#ifndef __IO_INTERFACE_HPP__
#define __IO_INTERFACE_HPP__


#include <iostream>
#include <memory>
#include <string>
#include <algorithm>
#include <boost/asio.hpp>
// #include <boost/iostreams/stream.hpp>
// #include <boost/iostreams/categories.hpp>
#include <boost/system/error_code.hpp>
#include <boost/system/system_error.hpp>
#include <boost/utility.hpp>
#include <boost/range/algorithm.hpp>

#include "throw_error.hpp"
#include "io_funcs.hpp"
#include "../utility/debug.hpp"



namespace orchid { namespace detail {


template <typename CO,typename Input>
class reader_basic
{
public:
    //////////////////////////////////////
    // typedef Coroutine coroutine_type;
    // typedef boost::shared_ptr<coroutine_type> coroutine_pointer;
    typedef CO coroutine_pointer;
    typedef char char_type;
    // typedef boost::iostreams::source_tag category;
public:
    reader_basic(Input& input,coroutine_pointer co):input_(input),co_(co) {

    }
    ~reader_basic() {
        
    }
public:
    //尝试读取size个字节到data中，返回值为实际读取的字节数，实际读取的字节数可能小于size
    //如果发生错误，被记录到e中
    std::size_t read(char* data,std::size_t size,boost::system::error_code& e) {
        return input_.read_some(boost::asio::buffer(data,size),co_,e);
    }
    //尝试读取size个字节到data中，返回值为实际读取的字节数，实际读取的字节数可能小于size
    //如果发生错误，抛出异常
    std::size_t read(char* data,std::size_t size) {
        return input_.read_some(boost::asio::buffer(data,size),co_);
    }

    //读取size个字节到data中，直到读满才返回。返回值为实际读取的字节数
    //如果发生错误，被记录到e中
    std::size_t read_full(char* data,std::size_t size,boost::system::error_code& e) {
        return orchid::read(input_,boost::asio::buffer(data,size),co_,e);
    }

    //读取size个字节到data中，直到读满才返回。返回值为实际读取的字节数
    //如果发生错误，抛出异常
    std::size_t read_full(char* data,std::size_t size) {
        return orchid::read(input_,boost::asio::buffer(data,size),co_);
    }

    //读取至少n个字节到data中。size为缓冲区长度，返回值为实际读取的字节数。
    //如果发生错误，被记录到e中
    std::size_t read_at_least(char* data,std::size_t size,std::size_t n,boost::system::error_code& e) {
        return orchid::read(input_,boost::asio::buffer(data,size),boost::asio::transfer_at_least(n),co_,e);
    }

    //读取至少n个字节到data中。size为缓冲区长度，返回值为实际读取的字节数。
    //如果发生错误，抛出异常
    std::size_t read_at_least(char* data,std::size_t size,std::size_t n) {
        return orchid::read(input_,boost::asio::buffer(data,size),boost::asio::transfer_at_least(n),co_);
    }

private:
    Input& input_;
    coroutine_pointer co_;
};



template <typename CO,typename Output>
class writer_basic
{
public:
    //////////////////////////////////////
    typedef CO coroutine_pointer;
    typedef char char_type;
    // typedef boost::iostreams::sink_tag category;
public:
    writer_basic(Output& output,coroutine_pointer co):output_(output),co_(co) {

    }
    ~writer_basic() {
        
    }
public:
    //尝试写入size个字节，返回值为实际写入的字节数，实际写入的字节数可能小于size
    //如果发生错误，被记录到e中
    std::size_t write(const char* data,std::size_t size,boost::system::error_code& e) {
        return output_.write_some(boost::asio::buffer(data,size),co_,e);
    }

    //尝试写入size个字节，返回值为实际写入的字节数，实际写入的字节数可能小于size
    //如果发生错误，抛出异常
    std::size_t write(const char* data,std::size_t size) {
        return output_.write_some(boost::asio::buffer(data,size),co_);
    }

    //写入size个字节，直到写完才返回，返回值为实际写入的字节数。
    //如果发生错误，被记录到e中
    std::size_t write_full(const char* data,std::size_t size,boost::system::error_code& e) {
        return orchid::write(output_,boost::asio::buffer(data,size),co_,e);
    }

    //写入size个字节，直到写完才返回，返回值为实际写入的字节数。
    //如果发生错误，抛出异常
    std::size_t write_full(const char* data,std::size_t size) {
        return orchid::write(output_,boost::asio::buffer(data,size),co_);
    }

    //写入至少n个字节。size为缓冲区长度。返回值为实际写入的字节数。
    //如果发生错误，被记录到e中
    std::size_t write_at_least(const char* data,std::size_t size,std::size_t n,boost::system::error_code& e) {
        return orchid::write(output_,boost::asio::buffer(data,size),boost::asio::transfer_at_least(n),co_,e);
    }

    //写入至少n个字节。size为缓冲区长度。返回值为实际写入的字节数。
    //如果发生错误，抛出异常
    std::size_t write_at_least(const char* data,std::size_t size,std::size_t n) {
        return orchid::write(output_,boost::asio::buffer(data,size),boost::asio::transfer_at_least(n),co_);
    }

private:
    Output& output_;
    coroutine_pointer co_;
};



template <typename CO,typename Input,typename Alloc=std::allocator<char> >
class buffered_reader_basic
{
public:
    //////////////////////////////////////
    // typedef Coroutine coroutine_type;
    // typedef boost::shared_ptr<coroutine_type> coroutine_pointer;
    typedef CO coroutine_pointer;
    typedef char char_type;
    enum {
        min_size = 16,
        default_size = 4096,
    };
    // typedef std::pair<char*,char*> range_type;
    // typedef boost::iostreams::source_tag category;
public:
    buffered_reader_basic(Input& input,coroutine_pointer co,std::size_t size = default_size)
        :input_(input),co_(co),w_(0),r_(0) {
            if(size < min_size) {
                size = min_size;
            }
            r_buff_.resize(size);
    }

    ~buffered_reader_basic() {
        
    }
public:
    //已缓存的字节数。
    std::size_t buffered() const {
        return w_ - r_;
    }
    //缓冲区的大小
    std::size_t size() const {
        return r_buff_.size();
    }
public:
    //尝试读取size个字节到data中，返回实际读取的字节数，实际读取的字节数可能少于size。
    //如果有错误，记录于e中。
    std::size_t read(char* data,std::size_t size,boost::system::error_code& e) {
        if (size <= 0 ) {
            return 0;
        }

        if ( buffered() == 0 ) {

            if (size >= this->size()) {
                return input_.read_some(boost::asio::buffer(data,size),co_,e);
            }

            fill(e);

            if (buffered() == 0) {
                return 0;
            }

        }

        if (size > buffered()) {
            size = buffered();
        } 

        std::copy(data,r_buff_.begin(),r_buff_.begin()+size);
        r_ += size;
        return size;

    }

    //尝试读取size个字节到data中，返回实际读取的字节数，实际读取的字节数可能少于size。
    //如果有错误，抛出异常
    std::size_t read(char* data,std::size_t size) {
        boost::system::error_code e;
        std::size_t bytes_transferred = read(data,size,e);
        if(e) {
            throw_error(e,"buffered io read");
        }
        return bytes_transferred;
    }


    //尝试读取数据到buf中，直到读取到delim为止，返回实际读取的字节数
    //如果有错误，记录在e中。
    template <typename Traits,typename Allocator>
    std::size_t read_until(std::basic_string<char,Traits,Allocator>& buf,char delim,boost::system::error_code& e) {

        buf.clear();
        int index = -1;
        std::size_t size = 0;
        int n = 0;

        index = find(r_buff_.begin()+r_,r_buff_.begin()+w_,delim);

        if (index != -1) {
            n = index+1;
            buf.reserve(n+1);
            std::copy(r_buff_.begin() + r_,r_buff_.begin()+ r_ + n,std::back_inserter(buf));
            r_ += n;
            size += n;
            return size;
        }
        // buf.reserve(this->size());
        
        for(;;) {

            n = buffered();
            if (n > 0) {
                std::copy(r_buff_.begin() + r_,r_buff_.begin() + w_,std::back_inserter(buf));
                r_ += n;
                size += n;
            }
            

            fill(e);

            if(e) {
                n = buffered();
                std::copy(r_buff_.begin() + r_,r_buff_.begin() + w_,std::back_inserter(buf));
                r_ += n;
                size += n;
                return size;
            }

            index = find(r_buff_.begin()+r_,r_buff_.begin()+w_,delim);

            if(index >= 0) {
                n = index+1;
                std::copy(r_buff_.begin() + r_,r_buff_.begin() + r_ + n,std::back_inserter(buf));
                r_ += n;
                size += n;
                return size;
            }

        }

    }

    //尝试读取数据到buf中，直到读取到delim为止，返回实际读取的字节数
    //如果有错误，抛出异常。
    template <typename Traits,typename Allocator>
    std::size_t read_until(std::basic_string<char,Traits,Allocator>& buf,char delim) {
        boost::system::error_code e;
        std::size_t bytes_transferred = read_until(buf,delim,e);
        if(e) {
            throw_error(e,"buffered io read_until");
        }
        return bytes_transferred;
    }


    //尝试读取数据到buf中，直到读取到delim为止，返回实际读取的字节数
    //如果有错误，记录在e中。
    template <typename Traits,typename Allocator>
    std::size_t read_until(std::basic_string<char,Traits,Allocator>& buf,const std::basic_string<char,Traits,Allocator>& delim,boost::system::error_code& e) {

        buf.clear();
        int index = -1;
        std::size_t size = 0;
        int n = 0;

        index = find(r_buff_.begin()+r_,r_buff_.begin()+w_,delim);

        if (index != -1) {
            n = index+delim.size();
            buf.reserve(n+1);
            std::copy(r_buff_.begin() + r_,r_buff_.begin()+ r_ + n,std::back_inserter(buf));
            r_ += n;
            size += n;
            return size;
        }

        // buf.reserve(this->size());
        
        for(;;) {

            n = buffered()-delim.size() + 1;
            if (n > 0) {
                std::copy(r_buff_.begin() + r_,r_buff_.begin() + r_ + n,std::back_inserter(buf));
                r_ += n;
                size += n;
            }
            
            fill(e);

            if(e) {
                n = buffered();
                std::copy(r_buff_.begin() + r_,r_buff_.begin() + r_ + n,std::back_inserter(buf));
                r_ += n;
                size += n;
                return size;
            }

            index = find(r_buff_.begin()+r_,r_buff_.begin()+w_,delim);

            if(index >= 0) {
                n = index + delim.size();
                std::copy(r_buff_.begin() + r_,r_buff_.begin() + r_ + n,std::back_inserter(buf));
                r_ += n;
                size += n;
                return size;
            }

        }

    }

    //尝试读取数据到buf中，直到读取到delim为止，返回实际读取的字节数
    //如果有错误，抛出异常。
    template <typename Traits,typename Allocator>
    std::size_t read_until(std::basic_string<char,Traits,Allocator>& buf,const std::basic_string<char,Traits,Allocator>& delim) {
        boost::system::error_code e;
        std::size_t bytes_transferred = read_until(buf,delim,e);
        if(e) {
            throw_error(e,"buffered io read_until");
        }
        return bytes_transferred;
    }


    //尝试读取数据到buf中，直到读取到delim为止，返回实际读取的字节数
    //如果有错误，记录在e中。
    template <typename Traits,typename Allocator>
    std::size_t read_until(std::basic_string<char,Traits,Allocator>& buf,const char* delim,boost::system::error_code& e) {
        return read_until(buf,std::string(delim),e);
    }


    //尝试读取数据到buf中，直到读取到delim为止，返回实际读取的字节数
    //如果有错误，抛出异常。
    template <typename Traits,typename Allocator>
    std::size_t read_until(std::basic_string<char,Traits,Allocator>& buf,const char* delim) {
        return read_until(buf,std::string(delim));
    }


    //读取size个字节到data中，直到读满为止，返回实际读取的字节数
    //如果有错误，记录在e中。
    std::size_t read_full(char* data,std::size_t size,boost::system::error_code& e) {
        std::size_t n = 0;
        std::size_t size_to_read = size;
        std::size_t size_read = 0;
        while (size_to_read) {
            n = read(data,size_to_read,e);
            size_to_read -= n;
            size_read += n;
            data += n;
            if(e) {
                break;
            }
        }
        return size_read;
    }

    //读取size个字节到data中，直到读满为止，返回实际读取的字节数
    //如果有错误，抛出异常。
    std::size_t read_full(char* data,std::size_t size) {
        boost::system::error_code e;
        std::size_t bytes_transferred = read_full(data,size,e);
        if(e) {
            throw_error(e,"buffered io read_full");
        }
        return bytes_transferred;
    }



private:
    void fill(boost::system::error_code& e) {
        if (r_ > 0) {
            std::memmove(&r_buff_[0],&r_buff_[r_],buffered());
            w_ -= r_;
            r_ = 0;
        }

        std::size_t n = input_.read_some(boost::asio::buffer(&r_buff_[w_], size() - buffered()), co_, e);
        w_ += n;

    }

    template <typename IT>
    int find(IT first, IT last,char delim) {
        for(std::size_t i = 0;first != last;++first,++i) {
            if(*first == delim) {
                return i;
            }
        }
        return -1;
    }

    template <typename IT,typename Traits,typename Allocator>
    int find(IT first, IT last,const std::basic_string<char,Traits,Allocator>& delim) {
        std::pair<IT,IT> range1(first,last);
        IT found = boost::search(range1,delim);
        if (found == last) {
            return -1;
        } else {
            return std::distance(first,found);
        }

    }

private:
    Input& input_;
    coroutine_pointer co_;
    std::vector<char,Alloc> r_buff_;
    std::size_t r_;
    std::size_t w_;
};




template <typename CO,typename Output,typename Alloc = std::allocator<char> >
class buffered_writer_basic
{
public:
    //////////////////////////////////////
    // typedef Coroutine coroutine_type;
    // typedef boost::shared_ptr<coroutine_type> coroutine_pointer;
    typedef CO coroutine_pointer;
    typedef char char_type;
    enum {
        min_size = 16,
        default_size = 4196,
    };
    // typedef boost::iostreams::sink_tag category;
public:
    buffered_writer_basic(Output& output,coroutine_pointer co,std::size_t size=default_size)
        :output_(output),co_(co),n_(0) {
        if(size < min_size) {
            size = min_size;
        }
        w_buff_.resize(size);

    }
    ~buffered_writer_basic() {

    }
public:
    //缓存的字节数
    std::size_t buffered() const {
        return n_;
    }
    //缓冲区大小
    std::size_t size() const {
        return w_buff_.size();
    }
    //可用空间大小
    std::size_t available() const {
        return size() - buffered();
    }

public:
    
    std::size_t write(const char* data,std::size_t size,boost::system::error_code& e) {
        std::size_t total = 0;
        while (size > available()) {
            std::size_t n = 0;
            if (buffered() == 0) {
                n = output_.write_some(boost::asio::buffer(data,size),co_,e);
            } else {
                n = available();
                std::copy(data,data+n,w_buff_.begin()+buffered());
                n_ += n;
                do_flush(e);
            }

            total += n;
            size -= n;
            data += n;

            if (e) {
                break;
            }
        }

        if (e) {
            return total;
        }

        std::copy(data,data+size,w_buff_.begin()+buffered());
        total += size;
        n_ += size;
        return total;
    }

    std::size_t write(const char* data,std::size_t size) {
        boost::system::error_code e;
        std::size_t n = write(data,size,e);
        if (e) {
            throw_error(e,"buffered io write");
        }
        return n;
    }

    void do_flush(boost::system::error_code& e) {
        if (buffered() == 0) {
            return;
        }

        std::size_t n = 0;
        n = output_.write_some(boost::asio::buffer(&w_buff_[0],buffered()),co_,e);

        if (n > 0 && n < buffered()) {
            std::memmove(&w_buff_[0],&w_buff_[n],buffered()-n);
        }

        n_ -= n;
    }

    void flush(boost::system::error_code& e) {
        while(buffered() > 0) {
            do_flush(e);
            if (e) {
                break;
            }
        }
    }

    void flush() {
        boost::system::error_code e;
        flush(e);
        if(e) {
            throw_error(e,"buffered io flush");
        }
    }

    // std::size_t write_full(const char* data,std::size_t size,boost::system::error_code& e) {
    //     std::size_t n = 0;
    //     std::size_t size_to_write = size;
    //     std::size_t size_wrote = 0;
    //     while (size_to_write) {
    //         n = write(data,size_to_write,e);
    //         size_to_write -= n;
    //         size_wrote += n;
    //         data += n;
    //         if(e) {
    //             break;
    //         }
    //     }
    //     return size_wrote;
    // }

    // std::size_t write_full(const char* data,std::size_t size) {
    //     boost::system::error_code e;
    //     write_full(data,size,e);
    //     if(e) {
    //         throw_error(e,"buffered io write_full");
    //     }
    // }

private:
    Output& output_;
    coroutine_pointer co_;
    // boost::asio::streambuf<Alloc> w_buff_;
    std::vector<char> w_buff_;
    std::size_t n_;

};


// template <typename T>
// class stream:public boost::iostreams::stream<T> {
// public:
//     template <typename IO,typename Coroutine>
//     stream(const IO& io,const Coroutine& co):boost::iostreams::stream<T>(io,co) {
//         this->exceptions(std::ios_base::badbit);
//     }

//     template <typename IO,typename Coroutine>
//     stream(IO& io,const Coroutine& co):boost::iostreams::stream<T>(io,co) {
//         this->exceptions(std::ios_base::badbit);
//     }

// };


}
}

#endif

<|start_filename|>unit_test/chat_client.cpp<|end_filename|>
// ====================================================================================
// Copyright (c) 2012, <EMAIL> All rights reserved.
// File         : chat_client.cpp
// Author       : <EMAIL>
// Last Change  : 11/28/2012 05:55 PM
// Description  : 
// ====================================================================================

#include <stdio.h>
#include <iostream>
#include <string>
#include <algorithm>
#include <iterator>
#include <vector>
#include <boost/scoped_ptr.hpp>
#include <boost/iostreams/stream.hpp>
#include <boost/algorithm/string.hpp>

#include "../orchid/all.hpp"

using std::string;
using std::cerr;
using std::endl;


class chat_client {
public:
    chat_client(orchid::scheduler& sche,const string& ip,const string& port)
        :sche_(sche),
        stdin_(sche_.get_io_service(),::dup(STDIN_FILENO)),
        stdout_(sche_.get_io_service(),::dup(STDOUT_FILENO)),
        sock_(sche_.get_io_service()),
        is_logined_(false),ip_(ip),port_(port) {

    }

    ~chat_client() {
        stdin_.close();
        stdout_.close();
        sock_.close();
    }
public:
    void run() {
        sche_.spawn(boost::bind(&chat_client::handle_console,this,_1));
        sche_.spawn(boost::bind(&chat_client::handle_msg,this,_1));
        sche_.run();
    }

    void stop() {
        sche_.stop();
    }

private:
    void handle_msg(orchid::coroutine_handle co) {
        try {
            sock_.connect(ip_,port_,co);
        } catch (const orchid::io_error& e) {
            ORCHID_ERROR("err msg:%s",e.what());
            return;
        }

        orchid::writer<orchid::descriptor> out(stdout_,co);
        orchid::buffered_reader<orchid::socket> in(sock_,co);

        string line;
        try {
            for (;;) {
                in.read_until(line,'\n');
                out.write_full(line.c_str(),line.size());
            }
        } catch (const orchid::io_error& e) {
            ORCHID_ERROR("err msg:%s",e.what());
        }
        
    }

    void handle_console(orchid::coroutine_handle co) {
        orchid::buffered_reader<orchid::descriptor> in(stdin_,co);
        orchid::buffered_writer<orchid::socket> out(sock_,co);

        string str;
        char buf[1024] = {0};
        try {
            for(;;) {
                in.read_until(str,'\n');
                ORCHID_DEBUG("read from stdin:%s",str.c_str());

                if(str.empty()) continue;
                // 退出 /q
                if(str.size() >= 2 && str[0] == '/' && str[1] == 'q') { 
                    sock_.close();
                    user_.clear();
                    is_logined_ = false;
                    ORCHID_ERROR("closed!");
                    stop();
                }
                // 发送消息 /s message
                else if(str.size() >= 4 && str[0] == '/' && str[1] =='s') {
                    if(!is_logined_) {
                        ORCHID_ERROR("please login first!");
                    } else {
                        int n = sprintf(buf,"%s:%s\r\n",user_.c_str(),str.c_str()+3);
                        out.write(buf,n);
                        out.flush();
                        ORCHID_DEBUG("send:%s n:%d",buf,n);
                    }
                }
                // 登陆 /l username 
                else if(str.size() >= 4 && str[0] == '/' && str[1] == 'l') {
                    if (!is_logined_) {
                        user_.assign(str.begin()+3,str.end()-1);
                        is_logined_ = true;
                        ORCHID_DEBUG("user:%s",user_.c_str());
                    } else {
                        ORCHID_ERROR("already logined!");
                    }
                } else {
                    print_err();
                }
            }
        } catch (const orchid::io_error& e) {
            ORCHID_ERROR("err:%s",e.what());
        }

    }

    void print_err() {
        ORCHID_ERROR("usage:\r\nlogin: /l name\r\nexit: /q\r\nsend: /s ooxx");
    }

private:
    orchid::scheduler& sche_;
    orchid::descriptor stdin_;
    orchid::descriptor stdout_;
    orchid::socket sock_;
    string user_;
    bool is_logined_;
    string ip_;
    string port_;

};

int main(int argc,char* argv[]) {
    orchid::scheduler sche;
    string ip = argv[1];
    string port = argv[2];
    chat_client client(sche,ip,port);
    client.run();
}

<|start_filename|>orchid/asio/signal.hpp<|end_filename|>
// ====================================================================================
// Copyright (c) 2012, <EMAIL> All rights reserved.
// File         : signal.hpp
// Author       : <EMAIL>
// Last Change  : 12/13/2012 11:47 AM
// Description  : 
// ====================================================================================


#ifndef __ORCHID_SIGNAL_H__
#define __ORCHID_SIGNAL_H__

#include <boost/asio.hpp>
#include <boost/utility.hpp>
#include <boost/system/error_code.hpp>
#include <boost/system/system_error.hpp>

#include "io_service.hpp"
#include "throw_error.hpp"
#include "handlers.hpp"
#include "../utility/debug.hpp"


namespace orchid { namespace detail{

class signal_basic:public boost::asio::signal_set {
public:
    typedef io_service io_service_type;
    // typedef signal_basic<Coroutine> self_type;
    // typedef Coroutine coroutine_type;
    // typedef typename coroutine_type::coroutine_pointer coroutine_pointer;



public:
    signal_basic(io_service_type& io_s)
        :boost::asio::signal_set(io_s) {

    }
    ~signal_basic() {

    }
public:

    template <typename CO>
    int wait(CO co,boost::system::error_code& e) {
        int sig_num = 0;
        sig_handler<CO> handler(co,e,sig_num);
        async_wait(handler);
        /////////////////////////
        co -> yield();
        ///////////////////////////
        if(e) {
            ORCHID_DEBUG("signal wait error: %s",e.message().c_str());
        }
        return sig_num;
    }

    template <typename CO>
    int wait(CO co) {
        boost::system::error_code e;
        int sig_num = wait(co,e);
        if(e) {
            // throw boost::system::system_error(e);
            throw_error(e,"signal wait");
        }
        return sig_num;
    }

};
    
}}



#endif

<|start_filename|>orchid/coroutine/coroutine.hpp<|end_filename|>
// ====================================================================================
// Copyright (c) 2012, <EMAIL> All rights reserved.
// File         : coroutine.hpp
// Author       : <EMAIL>
// Last Change  : 11/08/2012 12:34 AM
// Description  : 
// ====================================================================================

#ifndef __ORCHID_COROUTINE_H__
#define __ORCHID_COROUTINE_H__

#include <iostream>
#include <boost/assert.hpp>
#include <boost/context/all.hpp>
#include <boost/function.hpp>
#include <boost/bind.hpp>
#include <boost/utility.hpp>
#include <boost/atomic.hpp>
#include <boost/scoped_ptr.hpp>
#include <boost/enable_shared_from_this.hpp>

#include "stack_allocator.hpp"
#include "scheduler.hpp"
#include "../utility/debug.hpp"

namespace orchid { namespace detail {


class stack_unwind_exception {
public:
    stack_unwind_exception() {

    }
    ~stack_unwind_exception() {

    }
};


template <typename Scheduler,typename Alloc = stack_allocator>
class coroutine_basic
    :public boost::enable_shared_from_this<coroutine_basic<Scheduler,stack_allocator> >,
    boost::noncopyable {
public:
    typedef coroutine_basic<Scheduler,Alloc> self_type;
    typedef Alloc allocator_type;
    typedef Scheduler scheduler_type;
    typedef typename Scheduler::io_service_type io_service_type;
    typedef boost::context::fcontext_t context_type;
    typedef boost::shared_ptr<self_type> coroutine_handle;
    typedef boost::shared_ptr<self_type> coroutine_pointer;
    typedef boost::function<void(coroutine_handle)> func_type;

    static std::size_t default_stack_size() {
        return allocator_type::default_stack_size();
    }

    static std::size_t minimum_stack_size() {
        return allocator_type::minimum_stack_size();
    }

    static std::size_t maximum_stack_size() {
        return allocator_type::maximum_stack_size();
    }

    static void trampoline(intptr_t q) 
    {
        self_type* p = (self_type*)q;
        ORCHID_DEBUG("sche:%lu,coroutine:%lu,start running",p->sche_id(),p->id());
        try {
            if(p->is_stoped()) {
                throw stack_unwind_exception();
            }
            p -> f_(p -> get_handle());
        } catch(const stack_unwind_exception& e) {
            //do nothing.
            ORCHID_DEBUG("sche:%lu,coroutine:%lu,stack unwind",p->sche_id(),p->id());
        }
        p -> is_dead_.store(true);
        ORCHID_DEBUG("sche:%lu,coroutine:%lu,dead",p->sche_id(),p->id());
        boost::context::jump_fcontext(&p->ctx(),p->sche_.ctx(),0); //NOTITY SCHEDULER that coroutine is dead
    }

public:

    template <typename F>
    coroutine_basic(Scheduler& s,unsigned long id,const F& f,std::size_t stack_size)
		:id_(id),
		is_dead_(false),
        is_stoped_(false),
        alloc_(),
        f_(f), sche_(s) {
            ORCHID_DEBUG("sche:%lu,coroutine:%lu,coroutine_basic",this->sche_id(),this->id());
            stack_size_ = allocator_type::adjust_stack_size(stack_size);
            stack_pointer_ = alloc_.allocate(stack_size_);
            ctx_ = boost::context::make_fcontext(stack_pointer_,stack_size_,trampoline);
    }

    ~coroutine_basic() {
        ORCHID_DEBUG("sche:%lu,coroutine:%lu,~coroutine_basic()",sche_id(),id());
        alloc_.deallocate(stack_pointer_,stack_size_);
    }

public:
    /////////////////////////////接口函数////////////////////////////////////////

    //放弃运行，切换至调度器的上下文中
    //该函数只应该在当前协程的run函数中调用
    void yield() {
        ORCHID_DEBUG("sche:%lu,coroutine:%lu,yield",sche_id(),id());
        BOOST_ASSERT(!is_dead());
        boost::context::jump_fcontext(&ctx(),sche_.ctx(),0);
        if(is_stoped()) {
            //std::cout<<"throw here"<<std::endl;
            throw stack_unwind_exception();
        }
    }


    //恢复当前协程的上下文，立即切换
    void resume() {
        if(is_stoped() || is_dead()) {
            ORCHID_DEBUG("sche:%lu,coroutine:%lu,resume in coroutine,stop:%s dead:%s",
                sche_id(),id(),is_stoped()?"true":"false",is_dead()?"true":"false");
            return;
        }
        // ORCHID_DEBUG("sche:%lu,coroutine:%lu,resume",id(),sche_id());
        sche_.resume(this -> shared_from_this());
    }

    //恢复当前协程的上下文，但是并不是立即切换，
    //而是将切换的请求在调度器的IO_SERVICE里面排队，等待切换回上次运行的现场
    void sche_resume() {
        if(is_stoped() || is_dead()) {
            ORCHID_DEBUG("sche:%lu,coroutine:%lu,shce_resume in coroutine,stop:%s dead:%s",
                sche_id(),id(),is_stoped()?"true":"false",is_dead()?"true":"false");
            return;
        }
        // ORCHID_DEBUG("sche:%lu,coroutine:%lu,sche_resume",id(),sche_id());
        sche_.post(boost::bind(&scheduler_type::resume,&sche_,this->shared_from_this()));
    }

    //获取当前协程的运行状态。
    //该函数是线程安全的。
    bool is_dead() {
        return is_dead_.load();
    }


    //获取当前协程所在的调度器。
    scheduler_type& get_scheduler() {
        return sche_;
    }

    const scheduler_type& get_scheduler() const {
        return sche_;
    }

    coroutine_handle get_handle() {
        return this -> shared_from_this();
    }

    void stop() {
        is_stoped_.store(true);
    }

    bool is_stoped() {
        return is_stoped_.load();
    }

    std::size_t stack_size() {
        return stack_size_;
    }

    unsigned long id() const {
        return id_;
    }

    unsigned long sche_id() const {
        return sche_.id();
    }

    io_service_type& get_io_service() {
        return sche_.get_io_service();
    }

    const io_service_type& get_io_service() const {
        return sche_.get_io_service();
    }


    ///////////////////////////////内部函数//////////////////////////////////////

    context_type& ctx() {
        return ctx_;
    }

    const context_type& ctx() const {
        return ctx_;
    }

private:
    const unsigned long id_;
    boost::atomic_bool is_dead_;
    boost::atomic_bool is_stoped_;
    allocator_type alloc_;
    std::size_t stack_size_;
    void* stack_pointer_;
    context_type ctx_;
    func_type f_;
    scheduler_type& sche_;
};


} }


#endif


<|start_filename|>orchid/asio/handlers.hpp<|end_filename|>
// ====================================================================================
// Copyright (c) 2012, <EMAIL> All rights reserved.
// File         : handlers.hpp
// Author       : <EMAIL>
// Last Change  : 11/16/2012 05:58 PM
// Description  : 
// ====================================================================================
#ifndef __ORCHID_HANDLERS_H__
#define __ORCHID_HANDLERS_H__

#include <iostream>
#include <boost/asio.hpp>
#include <boost/utility.hpp>
#include <boost/system/error_code.hpp>
#include <boost/system/system_error.hpp>


namespace orchid { namespace detail {

template <typename CO>
struct io_handler {

    typedef CO coroutine_pointer;

    io_handler(coroutine_pointer co,boost::system::error_code& e,std::size_t& bytes_transferred):
        e_(e),co_(co),bytes_transferred_(bytes_transferred) {
    }

    void operator()(const boost::system::error_code& e,std::size_t bytes_transferred){
        e_ = e;
        bytes_transferred_ = bytes_transferred;
        co_ -> resume();
    }

    boost::system::error_code& e_;
    coroutine_pointer co_;
    std::size_t& bytes_transferred_;
};

template <typename CO>
struct connect_handler {

    typedef CO coroutine_pointer;

    connect_handler(coroutine_pointer co,
                boost::system::error_code& e,
                boost::asio::ip::tcp::resolver::iterator& it)
        :e_(e),co_(co),it_(it) {

    }

    void operator()(const boost::system::error_code& e,
            boost::asio::ip::tcp::resolver::iterator it) {
        it_ = it;
        e_ = e;
        co_ -> resume();
    }

    boost::system::error_code& e_;
    coroutine_pointer co_;
    boost::asio::ip::tcp::resolver::iterator& it_;

};

template <typename CO>
struct connect_without_resolver_handler {

	typedef CO coroutine_pointer;

	connect_without_resolver_handler(coroutine_pointer co,
		boost::system::error_code& e)
		:e_(e), co_(co) {

	}

	void operator()(const boost::system::error_code& e) {
		e_ = e;
		co_->resume();
	}

	boost::system::error_code& e_;
	coroutine_pointer co_;

};

template <typename CO>
struct sig_handler {

    typedef CO coroutine_pointer;

    sig_handler(coroutine_pointer co,boost::system::error_code& e,int& sig_num):
        e_(e),co_(co),sig_num_(sig_num) {
    }

    void operator()(const boost::system::error_code& e,int sig_num){
        e_ = e;
        sig_num_ = sig_num;
        co_ -> resume();
    }

    boost::system::error_code& e_;
    coroutine_pointer co_;
    int& sig_num_;
};

template <typename CO>
struct timer_handler {

    typedef CO coroutine_pointer;

    timer_handler(coroutine_pointer co,boost::system::error_code& e)
        :co_(co),e_(e) {

    }

    void operator()(const boost::system::error_code& e) {
        e_ = e;
        co_ -> resume();
    }

    coroutine_pointer co_;
    boost::system::error_code& e_;
};

template <typename CO>
struct accept_handler {

    typedef CO coroutine_pointer;

    accept_handler(coroutine_pointer co,boost::system::error_code& e):
        co_(co),e_(e) {
    }

    void operator()(const boost::system::error_code& e) {
        e_ = e;
        co_ -> resume();
    }

    coroutine_pointer co_;
    boost::system::error_code& e_;
};



template <typename CO>
struct worker_handler {
    typedef CO coroutine_pointer;

    worker_handler(coroutine_pointer co,const boost::function<void()>& f)
    :co_(co),f_(f){

    }

    void operator()() {
        f_();
        co_ -> sche_resume();
    }

    boost::function<void()> f_;
    coroutine_pointer co_;
};


}}

#endif




<|start_filename|>orchid/coroutine/stack_allocator.hpp<|end_filename|>
// ====================================================================================
// Copyright (c) 2012, <EMAIL> All rights reserved.
// File         : stack_allocator.hpp
// Author       : <EMAIL>
// Last Change  : 11/26/2012 04:38 PM
// Description  : 
// ====================================================================================
#ifndef __ORCHID_STACK_ALLOCATOR_H__
#define __ORCHID_STACK_ALLOCATOR_H__

#include <stdexcept>
#include <memory>
#include <unistd.h>
#include <stdlib.h>
#include <sys/mman.h>
#include <sys/resource.h>
#include <iostream>
#include <cmath>
#include <boost/assert.hpp>
#include <boost/context/all.hpp>
// #include <cstddef>
// #include <cstdlib>



namespace orchid { namespace detail {

class stack_allocator {
public:
    static std::size_t page_size() {
        return sysconf(_SC_PAGESIZE);
    }

    static std::size_t default_stack_size() {
        std::size_t size =  16*page_size();
        std::size_t max_size = maximum_stack_size();
        return size < max_size ? size:max_size;
    }

    static std::size_t maximum_stack_size() {
        struct rlimit rlim;
        BOOST_ASSERT(getrlimit(RLIMIT_STACK, &rlim) >= 0);
        return rlim.rlim_cur;
    }

    static std::size_t minimum_stack_size() {
        return 2*page_size();
    }

    static std::size_t adjust_stack_size(std::size_t stacksize) {
        std::size_t min = minimum_stack_size();
        std::size_t max = maximum_stack_size();
        BOOST_ASSERT(min<max);
        std::size_t pagesize = page_size();
        if (stacksize < min) {
            stacksize = min;
        }
        stacksize = stacksize + sizeof(boost::context::fcontext_t) + 15;
        stacksize = ((stacksize + pagesize- 1) & ~(pagesize - 1)) + pagesize;
        return stacksize < max ? stacksize : max;
    }
public:
    void* allocate(std::size_t stacksize) {
        void* p = ::mmap( 0, stacksize, PROT_READ | PROT_WRITE, MAP_PRIVATE | MAP_ANON, -1, 0);
        if(!p) {
            throw std::bad_alloc();
        }
        std::size_t pagesize = page_size();
        if (mprotect(p, pagesize, PROT_NONE) < 0) {
            munmap(p, stacksize);
            throw std::bad_alloc();
        }
        return static_cast<char*>(p) + stacksize;
    }

    void deallocate(void* p,std::size_t stacksize) {
        void * q = static_cast<char*>(p) - stacksize;
        munmap(q, stacksize);
    }

};
 
}}

#endif

<|start_filename|>unit_test/benchmark_orchid_client.cpp<|end_filename|>
// ====================================================================================
// Copyright (c) 2012, <EMAIL> All rights reserved.
// File         : benchmark_client.cpp
// Author       : <EMAIL>
// Last Change  : 11/22/2012 06:35 PM
// Description  : 
// ====================================================================================
#include <string>
#include <iostream>
#include <vector>

#include <boost/iostreams/stream.hpp>
#include <boost/chrono.hpp>
#include <boost/atomic.hpp>
#include <boost/lexical_cast.hpp>
#include <boost/scoped_array.hpp>
#include <boost/shared_ptr.hpp>

#include "../orchid/all.hpp"

using std::string;
using std::cout;
using std::endl;

const static std::size_t STACK_SIZE = 64*1024;

struct counter_t {
    long long total_read;
    long long total_write;
};

typedef boost::shared_ptr<orchid::socket> socket_ptr;

void handle_io(orchid::coroutine_handle co,orchid::chan<counter_t>& ch,
        socket_ptr sock,
        std::size_t buffer_size) {
    counter_t c;
    c.total_read = 0;
    c.total_write = 0;
    int n = buffer_size;
    try {
        boost::scoped_array<char> buff(new char[buffer_size]);
        sock -> connect("127.0.0.1","5678",co);
        sock -> set_option(boost::asio::ip::tcp::no_delay(true));
        printf("connect_sucess!!\r\n");
        for(;;) {
            n = sock -> write(buff.get(),n,co);
            c.total_write += n;
            n = sock -> read(buff.get(),buffer_size,co);
            c.total_read += n;
        }
    } catch (boost::system::system_error& e) {
        printf("errr:%s\r\n",e.what());
    }

    ch.send(c,co);
    printf("io_done!\r\n");

}


void controller(orchid::coroutine_handle co,
                    orchid::chan<counter_t>& ch,
                    std::size_t size,
                    std::size_t ms,
                    std::size_t buffer_size) {
        ////time start/////
    std::vector<socket_ptr> sockets;
    for(std::size_t i=0;i<size;++i) {
        socket_ptr sock(new orchid::socket(co -> get_scheduler().get_io_service()));
        sockets.push_back(sock);
        co -> get_scheduler().spawn(boost::bind(handle_io,_1,boost::ref(ch),sock,buffer_size), STACK_SIZE);
    }
    orchid::timer timer(co -> get_scheduler().get_io_service());
    timer.sleep(ms,co);
    for(std::size_t i=0;i<sockets.size();++i) {
        sockets[i] -> close();
        printf("lala_closing\r\n");
    }
}

void counter(orchid::coroutine_handle co,
            orchid::chan<counter_t>& ch,
            orchid::scheduler_group& group,
            std::size_t size) {
    counter_t c;
    counter_t tmp;
    c.total_read = 0;
    c.total_write = 0;
    //printf("total_%d\n",size);
    while(size-- > 0) {
        ch.recv(tmp,co);
        c.total_write += tmp.total_write;
        c.total_read += tmp.total_read;
    }
    cout<<"total_read:"<<c.total_read<<endl;
    cout<<"total_write:"<<c.total_write<<endl;
    group.stop();

}


int main(int arc,const char* argv[]) {
    int scheduler_count = boost::lexical_cast<int>(argv[1]);
    int session_count_per_scheduler = boost::lexical_cast<int>(argv[2]);
    int buffer_size = boost::lexical_cast<int>(argv[3]);
    int ms = boost::lexical_cast<int>(argv[4]) * 1000;

    cout<<"scheduler_count:"<<scheduler_count<<endl;
    cout<<"session_count_per_scheduler:"<<session_count_per_scheduler<<endl;

    orchid::scheduler_group sg(scheduler_count);
    int total_count = scheduler_count*session_count_per_scheduler;

    orchid::chan<counter_t> ch(total_count);
    sg[0].spawn(boost::bind(counter,_1,boost::ref(ch),boost::ref(sg),total_count), STACK_SIZE);

    for(std::size_t i=0;i<sg.size();++i) {
        sg[i].spawn(boost::bind(controller,_1,boost::ref(ch),session_count_per_scheduler,ms,buffer_size),
                    STACK_SIZE);
    }

    sg.run();
}

<|start_filename|>unit_test/test_client2.cpp<|end_filename|>
// ====================================================================================
// Copyright (c) 2012, <EMAIL> All rights reserved.
// File         : test_client.cpp
// Author       : <EMAIL>
// Last Change  : 11/17/2012 07:51 PM
// Description  : 
// ====================================================================================


#include <string>
#include <stdio.h>
#include <iostream>
#include <boost/iostreams/stream.hpp>
#include <boost/iostreams/copy.hpp>
#include <signal.h>

#include "../orchid/all.hpp"

using std::string;
using std::cout;
using std::cerr;
using std::endl;

void handle_io(orchid::coroutine_handle co) {
    orchid::socket sock_(co -> get_io_service());
    std::size_t n = 0;
    char buf[409600] = {0};

    try {
        sock_.connect("127.0.0.1","5678",co);
        ORCHID_DEBUG("id %lu connect success",co->id());
        orchid::writer<orchid::socket> writer(sock_,co);

        std::fill(buf,buf+409600,'a');
        for(int i = 0 ;i < 3;++ i) {
            n = writer.write_full(buf,409600);
            ORCHID_DEBUG("send %d bytes",n);
        }
        
        sock_.close();
 
    } catch (const orchid::io_error& e) {
        ORCHID_ERROR("id %lu msg:%s",co->id(),e.what());
        sock_.close();
    }
}

void handle_sig(orchid::coroutine_handle co) {
    orchid::signal sig(co -> get_io_service());
    try {
        sig.add(SIGINT);
        sig.add(SIGTERM);
        sig.wait(co);
        ORCHID_DEBUG("id %lu signal caught",co->id());
        co->get_scheduler().stop();

    } catch (const orchid::io_error& e) {
        ORCHID_ERROR("id %lu msg:%s",co->id(),e.what());
    }
}


int main() {
    orchid::scheduler sche;
    sche.spawn(handle_sig);
    for (int i=0;i<1;++i) {
        sche.spawn(handle_io,orchid::coroutine::maximum_stack_size());
    }
    sche.run();
}

<|start_filename|>unit_test/chat_server.cpp<|end_filename|>
// ====================================================================================
// Copyright (c) 2012, <EMAIL> All rights reserved.
// File         : chat_server.cpp
// Author       : <EMAIL>
// Last Change  : 11/30/2012 12:13 AM
// Description  : 
// ====================================================================================

#include <stdio.h>
#include <iostream>
#include <string>
#include <algorithm>
#include <iterator>
#include <vector>
#include <boost/bind.hpp>
#include <boost/iostreams/stream.hpp>
#include <boost/algorithm/string.hpp>
#include <boost/variant.hpp>
#include <boost/thread.hpp>

#include "../orchid/all.hpp"

using std::string;
using std::cerr;
using std::cout;
using std::endl;

const static std::size_t STACK_SIZE = 64*1024;

template <typename Client>
struct server {
    typedef server<Client> self_type;
    typedef boost::shared_ptr<Client> client_sp_type;
    typedef std::list<client_sp_type> client_list_type;

    enum {REGISTER,UNREGISTER};
    struct ctrl_t {
        int cmd_;
        client_sp_type client_;
    };
    typedef boost::variant<string,ctrl_t> msg_type;

    ///////////////////////
    orchid::scheduler& main_sche_;
    orchid::scheduler_group& schedulers_;
    orchid::acceptor acceptor_;
    std::string port_;
    orchid::chan<msg_type> msg_ch_;
    client_list_type clients_;
    ///////////////////////////

    server(orchid::scheduler& sche,orchid::scheduler_group& s,const string& port)
        :main_sche_(sche),
        schedulers_(s),
        acceptor_(main_sche_.get_io_service()),
        port_(port),msg_ch_(1) {
    }
    ~server() {
        acceptor_.close();
    }

    //主服务协程，不停的从chan里面的读取消息，并处理消息。
    void process_msg(orchid::coroutine_handle co) {
        for (;;) {
            msg_type msg;
            if (!msg_ch_.recv(msg,co)) {
                ORCHID_DEBUG("chan closed");
                break;
            }

            switch (msg.which()) {
                case 0:
                    ORCHID_DEBUG("process msg:%s",boost::get<string>(msg).c_str());
                    for(typename client_list_type::iterator it = clients_.begin(); it != clients_.end(); ++it) {
                        (*it) -> ch_.send(boost::get<string>(msg),co);
                    }
                    break;
                case 1:
                    if(boost::get<ctrl_t>(msg).cmd_ == REGISTER) {
                        ORCHID_DEBUG("on register");
                        boost::get<ctrl_t>(msg).client_->start();
                        clients_.push_back(boost::get<ctrl_t>(msg).client_);
                    } else if(boost::get<ctrl_t>(msg).cmd_ == UNREGISTER) {
                        ORCHID_DEBUG("on unregister");
                        boost::get<ctrl_t>(msg).client_->close();
                        clients_.remove(boost::get<ctrl_t>(msg).client_);
                    } else {
                        throw std::runtime_error("unkonw cmd! should never hanppened!");
                    }
                    break;
                default:
                    throw std::runtime_error("unkonw msg! should never hanppened!");
            }
        }
    }

    void accept(orchid::coroutine_handle co) {
        try {
            int index = 0;
            acceptor_.bind_and_listen(port_,true);
            for (;;) {
                if(index >= schedulers_.size()) index = 0;
                boost::shared_ptr<Client> c(new Client(schedulers_[index++],*this));
                acceptor_.accept(c->sock_,co);
                ORCHID_DEBUG("on accept");
                ctrl_t msg;
                msg.cmd_ = REGISTER;
                msg.client_ = c;
                msg_ch_.send(msg,co);
            }
        } catch (boost::system::system_error& e) {
            ORCHID_ERROR("err:%s",e.what());
        }
    }

    void run() {
        main_sche_.spawn(boost::bind(&self_type::accept,this,_1));
        main_sche_.spawn(boost::bind(&self_type::process_msg,this,_1));
        boost::thread t(boost::bind(&orchid::scheduler_group::run,&schedulers_));
        main_sche_.run();
        t.join();
    }

};


struct client_agent:public boost::enable_shared_from_this<client_agent> {
    orchid::scheduler& sche_;
    server<client_agent>& server_;
    orchid::socket sock_;
    orchid::chan<string> ch_;

    client_agent(orchid::scheduler& sche,server<client_agent>& s)
        :sche_(sche),server_(s),
        sock_(sche_.get_io_service()),ch_(32) {

    }
    ~client_agent() {
        ORCHID_DEBUG("~client_agent()");
        sock_.close();
    }

    void start() {
         sche_.spawn(boost::bind(&client_agent::sender,this -> shared_from_this(),_1));
         sche_.spawn(boost::bind(&client_agent::receiver,this -> shared_from_this(),_1));
    }

    void close() {
        ch_.close();
    }

    void sender(orchid::coroutine_handle co) {
        string str;
        orchid::buffered_writer<orchid::socket> out(sock_,co);
        try {
            while(ch_.recv(str,co)) {
                out.write(str.c_str(),str.size());
                out.flush();
            }
        } catch(const orchid::io_error& e) {
            ORCHID_ERROR("err:%s",e.what());
        }
        ORCHID_DEBUG("sender: exit!");
    }

    void receiver(orchid::coroutine_handle co) {
        orchid::buffered_reader<orchid::socket> in(sock_,co);
        string str;
        try {
            for (;;) {
                in.read_until(str,'\n');
                ORCHID_DEBUG("recv:%s",str.c_str());
                server_.msg_ch_.send(str,co);
            }
        } catch(const orchid::io_error& e) {
            ORCHID_ERROR("err:%s",e.what());
        }

        server<client_agent>::ctrl_t ctrl_msg;
        ctrl_msg.cmd_ = server<client_agent>::UNREGISTER;
        ctrl_msg.client_ = this -> shared_from_this();
        server_.msg_ch_.send(ctrl_msg, co);
        ORCHID_DEBUG("receiver: exit!");
    }
};

int main(int argc,char* argv[]) {
    string port = argv[1];
    orchid::scheduler main_sche;
    orchid::scheduler_group group(4);
    server<client_agent> s(main_sche,group,port);
    s.run();

}

<|start_filename|>orchid/asio/descriptor.hpp<|end_filename|>
// ====================================================================================
// Copyright (c) 2012, <EMAIL> All rights reserved.
// File         : stream_descriptor.hpp
// Author       : <EMAIL>
// Last Change  : 11/27/2012 10:15 PM
// Description  : 
// ====================================================================================
#ifndef __ORCHID_STREAM_DESCRIPTOR_H__
#define __ORCHID_STREAM_DESCRIPTOR_H__

#include <iostream>
#include <boost/asio.hpp>
#include <boost/utility.hpp>
#include <boost/system/error_code.hpp>
#include <boost/system/system_error.hpp>

#include "io_service.hpp"
#include "throw_error.hpp"
#include "handlers.hpp"
#include "../utility/debug.hpp"

namespace orchid { namespace detail {

class descriptor_basic:public boost::asio::posix::stream_descriptor {
public:
    typedef boost::asio::posix::stream_descriptor::native_handle_type native_handle_type;
    typedef io_service io_service_type;
    // typedef descriptor_basic<Coroutine> self_type;
    // typedef Coroutine coroutine_type;
    // typedef typename coroutine_type::coroutine_pointer coroutine_pointer;

public:
    descriptor_basic(io_service_type& io_s)
        :boost::asio::posix::stream_descriptor(io_s) {

    }


    descriptor_basic(io_service_type& io_s,const native_handle_type& handle)
        :boost::asio::posix::stream_descriptor(io_s,handle) {

    }


    ~descriptor_basic() {
        
    }
    
public:


    template <typename MutableBufferSequence,typename CO>
    std::size_t read_some(const MutableBufferSequence& buf,CO co,boost::system::error_code& e) {
        BOOST_ASSERT(co != NULL);
        std::size_t bytes_transferred;
        io_handler<CO> handler(co,e,bytes_transferred);
        async_read_some(buf, handler);
        //////////////////////////////////
        co->yield();
        //////////////////////////////////
        if (e) {
            ORCHID_DEBUG("description read error: %s",e.message().c_str());
        }
        return bytes_transferred; 
    }

    template <typename MutableBufferSequence,typename CO>
    std::size_t read_some(const MutableBufferSequence& buf,CO co) {
        BOOST_ASSERT(co != NULL);
        boost::system::error_code e;
        std::size_t bytes_transferred = read_some(buf,co,e);

        if (e) {
            // throw boost::system::system_error(e);
            throw_error(e,"descriptor read");
        }

        return bytes_transferred; 
    }


    template <typename ConstBufferSequence,typename CO>
    std::size_t write_some(const ConstBufferSequence& buf,CO co,boost::system::error_code& e) {
        BOOST_ASSERT(co != NULL);
        std::size_t bytes_transferred;
        io_handler<CO> handler(co,e,bytes_transferred);
        async_write_some(buf, handler);
        //////////////////////////////////
        co->yield();
        //////////////////////////////////
        if (e) {
            ORCHID_DEBUG("description write error: %s",e.message().c_str());
        }
        return bytes_transferred; 
    }


    template <typename ConstBufferSequence,typename CO>
    std::size_t write_some(const ConstBufferSequence& buf,CO co) {
        BOOST_ASSERT(co != NULL);
        boost::system::error_code e;
        std::size_t bytes_transferred = write_some(buf,co,e);
        if (e) {
            // throw boost::system::system_error(e);
            throw_error(e,"descriptor write");
        }

        return bytes_transferred; 
    }



};


}}


#endif

<|start_filename|>orchid/utility/debug.hpp<|end_filename|>

#ifndef __ORCHID_DEBUG_H__
#define __ORCHID_DEBUG_H__

#include <stdio.h>  
  
// #define __ORCHID_DEBUG__
  
#ifdef __ORCHID_DEBUG__ 
#define ORCHID_DEBUG(format,...)  printf("File: " __FILE__", Line: %05d: " format"\n", __LINE__, ##__VA_ARGS__)
#else  
#define ORCHID_DEBUG(format,...)  
#endif

#define __ORCHID_ERROR__

#ifdef __ORCHID_ERROR__ 
#define ORCHID_ERROR(format,...)  fprintf(stderr,"File: " __FILE__", Line: %05d: " format"\n", __LINE__, ##__VA_ARGS__)
#else  
#define ORCHID_ERROR(format,...)  
#endif


#endif

<|start_filename|>orchid/asio/worker.hpp<|end_filename|>
// ====================================================================================
// Copyright (c) 2012, <EMAIL> All rights reserved.
// File         : worker.hpp
// Author       : <EMAIL>
// Last Change  : 11/19/2012 04:15 PM
// Description  : 
// ====================================================================================


#include <boost/asio.hpp>
#include <boost/assert.hpp>
#include <boost/function.hpp>
#include <boost/bind.hpp>
#include <boost/thread.hpp>

#include "io_service.hpp"
#include "throw_error.hpp"
#include "handlers.hpp"
#include "../utility/debug.hpp"

namespace orchid {

    template <typename CO>
    void run_in_thread(const boost::function<void()>& f,CO co) {
        detail::worker_handler<CO> handler(co,f);
        boost::thread t(handler);
        co -> yield();
    }

namespace detail {

    class worker_pool_basic {
    public:
        typedef io_service io_service_type;

    public:
        worker_pool_basic(std::size_t size)
            :size_(size),io_service_() {

        }
        ~worker_pool_basic() {

        }
    public:
        void run() {
            for(std::size_t i = 0;i < size_;++i) {
                tg_.create_thread(boost::bind(&io_service_type::run,&io_service_));
            }
            tg_.join_all();
        }

        void stop() {
            io_service_.stop();
        }

        template <typename CO>
        void post(const boost::function<void()>& f,CO co) {
            detail::worker_handler<CO> handler(co,f);
            io_service_.post(handler);
            co -> yield();
        } 

        std::size_t size() {
            return size_;
        }

    private:

        std::size_t size_;
        boost::thread_group tg_;
        io_service io_service_;

    };


}


}

<|start_filename|>orchid/asio/throw_error.hpp<|end_filename|>

#ifndef __ORCHID_THROW_EXCEPTION_H__
#define __ORCHID_THROW_EXCEPTION_H__

#include <boost/system/error_code.hpp>
#include <boost/system/system_error.hpp>
#include <iostream>
#include <boost/asio.hpp>

namespace orchid { namespace detail {

    class io_error: public boost::system::system_error {
        
    public:
        explicit io_error(const boost::system::error_code& err)
            :boost::system::system_error(err) {

        }

        io_error(const boost::system::error_code& err,const char* location)
            :boost::system::system_error(err,location) {

        }

    };

    static void throw_error(const boost::system::error_code& err) {
      io_error e(err);
      throw e;
    }

	static void throw_error(const boost::system::error_code& err, const char* location) {
      io_error e(err, location);
      throw e;
    }
}}


#endif

<|start_filename|>orchid/asio/socket.hpp<|end_filename|>
// ====================================================================================
// Copyright (c) 2012, <EMAIL> All rights reserved.
// File         : socket.hpp
// Author       : <EMAIL>
// Last Change  : 11/13/2012 04:04 PM
// Description  : 
// ====================================================================================

#ifndef __ORCHID_SOCKET_H__
#define __ORCHID_SOCKET_H__

#include <string>
#include <iostream>

#include <boost/assert.hpp>
#include <boost/asio.hpp>
#include <boost/utility.hpp>
#include <boost/system/error_code.hpp>
#include <boost/system/system_error.hpp>

#include "throw_error.hpp"
#include "io_service.hpp"
#include "handlers.hpp"
#include "../utility/debug.hpp"

namespace orchid { namespace detail {


class socket_basic: public boost::asio::ip::tcp::socket
{
public:
    ////////////////////////////////////////
    typedef io_service io_service_type;
    // typedef socket_basic<Coroutine> self_type;
    // typedef Coroutine coroutine_type;
    // typedef typename coroutine_type::coroutine_pointer coroutine_pointer;

public:
    socket_basic(io_service_type& io_s):boost::asio::ip::tcp::socket(io_s) {

    }

    ~socket_basic() {

    }
public:

	template <typename CO>
	void connect(const boost::asio::ip::tcp::endpoint& addr, CO co, boost::system::error_code& e) {
		BOOST_ASSERT(co != NULL);
		/////////////////////////////////////////
		connect_without_resolver_handler<CO> c_handler(co, e);
		async_connect(addr, c_handler);
		co->yield();
		////////////////////////////////////
		if (e) {
			ORCHID_DEBUG("socket connect error: %s", e.message().c_str());
		}
		return;
	}


	template <typename CO>
	void connect(const boost::asio::ip::tcp::endpoint& addr, CO co) {
		BOOST_ASSERT(co != NULL);
		boost::system::error_code e;
		connect(addr, co, e);
		if (e) {
			throw_error(e, "socket connect");
		}
	}

    template <typename CO>
    void connect(const std::string& host,const std::string& port,CO co,boost::system::error_code& e) {
        BOOST_ASSERT(co != NULL);
        boost::asio::ip::tcp::resolver resolver(get_io_service());
        boost::asio::ip::tcp::resolver::iterator start;

        connect_handler<CO> r_handler(co,e,start);
        boost::asio::ip::tcp::resolver::query query(host,port);
        resolver.async_resolve(query, r_handler);
        //start = resolver.resolve(query);
        ///////////////////////////////////////////
        co -> yield();
        ////////////////////////////////////////////
        if(e) {
            ORCHID_DEBUG("socket connect error: %s",e.message().c_str());
            return;
        }
        /////////////////////////////////////////
        connect_handler<CO> c_handler(co,e,start);
        boost::asio::async_connect(*this,start, c_handler);
        co -> yield();
        ////////////////////////////////////
        if(e) {
            ORCHID_DEBUG("socket connect error: %s",e.message().c_str());
        }
        return;
    }

    template <typename CO>
    void connect(const std::string& host,const std::string& port,CO co) {
        BOOST_ASSERT(co != NULL);
        boost::system::error_code e;
        connect(host,port,co,e);
        if(e) {
            throw_error(e,"socket connect");
        }
    }

    template <typename MutableBufferSequence,typename CO>
    std::size_t read_some(const MutableBufferSequence& buf,CO co,boost::system::error_code& e) {
        BOOST_ASSERT(co != NULL);
        std::size_t bytes_transferred;
        io_handler<CO> handler(co,e,bytes_transferred);
        async_read_some(buf, handler);
        //////////////////////////////////
        co->yield();
        //////////////////////////////////
        if(e) {
            ORCHID_DEBUG("socket read error: %s",e.message().c_str());
        }
        return bytes_transferred;
    }


    template <typename MutableBufferSequence,typename CO>
    std::size_t read_some(const MutableBufferSequence& buf,CO co) {
        BOOST_ASSERT(co != NULL);
        boost::system::error_code e;
        std::size_t bytes_transferred = read_some(buf,co,e);
        //////////////////////////////////
        if (e) {
            // throw boost::system::system_error(e);
            throw_error(e,"socket read");
        }
        return bytes_transferred; 
    }

    template <typename ConstBufferSequence,typename CO>
    std::size_t write_some(const ConstBufferSequence& buf,CO co,boost::system::error_code& e) {
        BOOST_ASSERT(co != NULL);
        std::size_t bytes_transferred;
        io_handler<CO> handler(co,e,bytes_transferred);
        async_write_some(buf, handler);
        //////////////////////////////////
        co->yield();
        //////////////////////////////////
        if(e) {
            ORCHID_DEBUG("socket write error: %s",e.message().c_str());
        }
        return bytes_transferred; 
    }


    template <typename ConstBufferSequence,typename CO>
    std::size_t write_some(const ConstBufferSequence& buf,CO co) {
        BOOST_ASSERT(co != NULL);
        boost::system::error_code e;
        std::size_t bytes_transferred = write_some(buf,co,e);
        if (e) {
            // throw boost::system::system_error(e);
            throw_error(e,"socket write");
        }
        return bytes_transferred;
    }



};


}
}



#endif

<|start_filename|>orchid/coroutine/chan.hpp<|end_filename|>
// ====================================================================================
// Copyright (c) 2012, <EMAIL> All rights reserved.
// File         : chan.hpp
// Author       : <EMAIL>
// Last Change  : 11/18/2012 01:35 AM
// Description  : 
// ====================================================================================

#ifndef __ORCHID_CHAN_H__
#define __ORCHID_CHAN_H__

#include <iostream>
#include <memory>
#include <queue>

#include <boost/utility.hpp>
#include <boost/shared_ptr.hpp>
#include <boost/scoped_array.hpp>

#include "spinlock.hpp"


namespace orchid { namespace detail {

////////////////////chan类型模拟了GOLANG中chan的概念/////////////////


template <typename CO,typename T>
class chan_basic:boost::noncopyable 
{
public:
    typedef chan_basic<CO,T> self_type;
    typedef T value_type;
    typedef CO coroutine_pointer;
public:
    chan_basic(std::size_t cap)
		:cap_(cap), size_(0),array_(new T[cap]), r_(0),
        w_(0),is_closed_(false) {
    }
    ~chan_basic() {
        close(); 
    }
public:

    void close() {
        locker_.lock();
        if (is_closed_) {
            locker_.unlock();
            return;
        }

        is_closed_ = true;
        while(!r_queue_.empty()){
            r_queue_.front() -> sche_resume();
            r_queue_.pop();
        }
        while(!w_queue_.empty()){
            w_queue_.front() -> sche_resume();
            w_queue_.pop();
        }
        locker_.unlock();
    }

    template <typename U>
    bool send(const U& t,coroutine_pointer co) {
        locker_.lock();
        if(is_closed_) {
            locker_.unlock();
            return false;
        }

        while(size_ >= cap_) {
            w_queue_.push(co);
            locker_.unlock();
            co -> yield();
            locker_.lock();
            if(is_closed_) {
                locker_.unlock();
                return false;
            }
        }

        array_[w_++] = t;
        if(w_ == cap_) w_ = 0;
        ++size_;
        if(!r_queue_.empty()) {
            coroutine_pointer r = r_queue_.front();
            r_queue_.pop();
            locker_.unlock();
            r -> sche_resume();
        } else {
            locker_.unlock();
        }
        // co -> sche_resume();
        // co -> yield();
        return true;
    }

    template <typename U>
    bool recv(U& t,coroutine_pointer co) {
        locker_.lock();
        while(size_ <= 0) {

            if(is_closed_) {
                locker_.unlock();
                return false;
            }

            r_queue_.push(co);
            locker_.unlock();
            co -> yield();
            locker_.lock();
 
        }

        std::swap(t,array_[r_++]);
        //t = array_[r++];
        if(r_ == cap_) r_ = 0;
        --size_;

        if(!w_queue_.empty()) {
            coroutine_pointer w = w_queue_.front();
            w_queue_.pop();
            locker_.unlock();
            w -> sche_resume();
        } else {
            locker_.unlock();
        }

        return true;
    }



private:
    spinlock locker_;
    std::size_t cap_;
    int size_;
    boost::scoped_array<T> array_;
    std::size_t r_;
    std::size_t w_;
    bool is_closed_;
    std::queue<coroutine_pointer> r_queue_;
    std::queue<coroutine_pointer> w_queue_;
    
};



}
}

#endif

<|start_filename|>orchid/coroutine/spinlock.hpp<|end_filename|>
// ====================================================================================
// Copyright (c) 2012, <EMAIL> All rights reserved.
// File         : spinlock.hpp
// Author       : <EMAIL>
// Last Change  : 12/01/2012 09:14 PM
// Description  : 
// ====================================================================================


#ifndef __ORCHID_SPINLOCK_H__
#define __ORCHID_SPINLOCK_H__
#include <boost/smart_ptr/detail/spinlock.hpp>
namespace orchid {namespace detail {
class spinlock:boost::noncopyable {
private:
	boost::detail::spinlock lock_;
public:
	spinlock(){ lock_ = BOOST_DETAIL_SPINLOCK_INIT; }
  
  void lock()
  {
	  lock_.lock();
  }
  void unlock()
  {
	  lock_.unlock();
  }
};


}}


#endif

<|start_filename|>orchid/asio/timer.hpp<|end_filename|>
// ====================================================================================
// Copyright (c) 2012, <EMAIL> All rights reserved.
// File         : timer.hpp
// Author       : <EMAIL>
// Last Change  : 11/19/2012 04:15 PM
// Description  : 
// ====================================================================================


#include <iostream>
#include <boost/asio.hpp>
#include <boost/assert.hpp>
#include <boost/system/error_code.hpp>
#include <boost/system/system_error.hpp>

#include "io_service.hpp"
#include "throw_error.hpp"
#include "handlers.hpp"
#include "../utility/debug.hpp"

namespace orchid { namespace detail {


class timer_basic:public boost::asio::deadline_timer {
public:
    typedef io_service io_service_type;
    // typedef timer_basic<Coroutine> self_type;
    // typedef Coroutine coroutine_type;
    // typedef typename coroutine_type::coroutine_pointer coroutine_pointer;


public:
    timer_basic(io_service_type& io_s):boost::asio::deadline_timer(io_s) {

    }

    ~timer_basic() {

    }
public:

    template <typename CO>
    void sleep(std::size_t ms,CO co,boost::system::error_code& e) {
        BOOST_ASSERT(co != NULL);
        if (ms == 0) {
            return;
        }
        timer_handler<CO> handler(co,e);
        expires_from_now(boost::posix_time::milliseconds(ms));
        async_wait(handler);
        //////////////////////////
        co -> yield();
        /////////////////////////
        if(e) {
            ORCHID_DEBUG("timer sleep error: %s",e.message().c_str());
        }        
        return;
    }

    template <typename CO>
    void sleep(std::size_t ms,CO co) {
        BOOST_ASSERT(co != NULL);
        boost::system::error_code e;
        sleep(ms,co,e);
        if(e) {
            // throw boost::system::system_error(e);
            throw_error(e,"timer wait");
        }
    }


};






} }

<|start_filename|>unit_test/benchmark_orchid_server.cpp<|end_filename|>
// ====================================================================================
// Copyright (c) 2012, <EMAIL> All rights reserved.
// File         : orchid_benchmark.cpp
// Author       : <EMAIL>
// Last Change  : 11/22/2012 06:28 PM
// Description  : 
// ====================================================================================

#include <string>
#include <iostream>
#include <boost/iostreams/stream.hpp>
#include <boost/lexical_cast.hpp>
#include <boost/scoped_array.hpp>
#include <boost/shared_ptr.hpp>

#include "../orchid/all.hpp"

using std::string;
using std::cout;
using std::endl;

const static std::size_t STACK_SIZE = 64*1024;

typedef boost::shared_ptr<orchid::socket> socket_ptr;

void handle_io(orchid::coroutine_handle co,socket_ptr sock,std::size_t buffer_size) {
    try {
        boost::scoped_array<char> buffer(new char[buffer_size]);
        sock -> set_option(boost::asio::ip::tcp::no_delay(true));
        printf("accept_sucess!!\r\n");
        int n = 0;
        for (;;) {
            n = sock -> read(buffer.get(),buffer_size,co);
            n = sock -> write(buffer.get(),n,co);
        }

    } catch (boost::system::system_error& e) {
        cout<<e.code()<<" "<<e.what()<<endl;
    }
}

void handle_accept(orchid::coroutine_handle co,
    orchid::scheduler_group& group,
    const std::string& port,
    std::size_t buffer_size) {
    try {
        orchid::acceptor acceptor(co -> get_scheduler().get_io_service());
        acceptor.bind_and_listen(port,true);
        int i = 0;
        for(;;) {
            socket_ptr sock(new orchid::socket(group[i].get_io_service()));
            acceptor.accept(*sock,co);
            group[i].spawn(boost::bind(handle_io,_1,sock,buffer_size),STACK_SIZE);
            if(++i == group.size()) i=0;
        }
    }
    catch(boost::system::system_error& e) {
        cout<<e.code()<<" "<<e.what()<<endl;
        cout<<"acceptor's done!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!"<<endl;
    }
}

int main(int argc,const char* argv[]) {
    std::string port = "5678";
    int thread_count = boost::lexical_cast<int>(argv[1]);
    int buf_size = boost::lexical_cast<int>(argv[2]);
    cout<<"thread_count:"<<thread_count<<endl;
    orchid::scheduler_group group(thread_count);
    
    group[0].spawn(boost::bind(handle_accept,_1,boost::ref(group),boost::cref(port),buf_size),
                    STACK_SIZE);
    group.run();
}

<|start_filename|>orchid/asio/acceptor.hpp<|end_filename|>
// ====================================================================================
// Copyright (c) 2012, <EMAIL> All rights reserved.
// File         : acceptor.hpp
// Author       : <EMAIL>
// Last Change  : 11/16/2012 05:58 PM
// Description  : 
// ====================================================================================

#ifndef __ORCHID_ACCEPTOR_H__
#define __ORCHID_ACCEPTOR_H__
#include <iostream>
#include <string>
#include <boost/asio.hpp>
#include <boost/assert.hpp>
#include <boost/system/error_code.hpp>
#include <boost/system/system_error.hpp>
// #include <boost/lexical_cast.hpp>

#include "io_service.hpp"
#include "socket.hpp"
#include "throw_error.hpp"
#include "handlers.hpp"
#include "../utility/debug.hpp"


namespace orchid { namespace detail {

class acceptor_basic:public boost::asio::ip::tcp::acceptor {
public:
    typedef socket_basic socket_type;
    typedef io_service io_service_type;
    // typedef acceptor_basic<Coroutine> self_type;
    // typedef Coroutine coroutine_type;
    // typedef typename coroutine_type::coroutine_pointer coroutine_pointer;


public:
    acceptor_basic(io_service_type& io_s):boost::asio::ip::tcp::acceptor(io_s) {

    }

    ~acceptor_basic() {
        
    }
public:
    void bind_and_listen(const std::string& port,bool reuse_addr) {
        boost::asio::ip::tcp::resolver resolver(get_io_service());
        boost::asio::ip::tcp::resolver::query query(boost::asio::ip::tcp::v4(),port);
        boost::asio::ip::tcp::endpoint endpoint = *resolver.resolve(query);
        open(endpoint.protocol());
        if(reuse_addr) {
            set_option(boost::asio::ip::tcp::acceptor::reuse_address(true));
        }
        bind(endpoint);
        listen(boost::asio::socket_base::max_connections);
    }

    void bind_and_listen(const std::string& port,bool reuse_addr,boost::system::error_code& e) {
        try {
            bind_and_listen(port,reuse_addr);
        } catch (const boost::system::system_error& ex) {
            e = ex.code();
            ORCHID_DEBUG("bind_and_listen error: %s",e.message().c_str());
        }
        
    }

	void bind_and_listen(const boost::asio::ip::tcp::endpoint& addr, bool reuse_addr) {
		open(addr.protocol());
		if (reuse_addr)
			set_option(boost::asio::ip::tcp::acceptor::reuse_address(true));
		bind(addr);
		listen(boost::asio::socket_base::max_connections);
	}

	void bind_and_listen(const boost::asio::ip::tcp::endpoint& addr, bool reuse_addr, boost::system::error_code& e) {
		try {
			bind_and_listen(addr, reuse_addr);
		}
		catch (const boost::system::system_error& ex) {
			e = ex.code();
			ORCHID_DEBUG("bind_and_listen error: %s", e.message().c_str());
		}
	}

    template <typename CO>
    void accept(socket_type& sock,CO co,boost::system::error_code& e) {
        BOOST_ASSERT(co != NULL);
        accept_handler<CO> handler(co,e);
        async_accept(sock, handler);
        ///////////////////////////////////
        co -> yield();
        ////////////////////////////////////
        if(e) {
            ORCHID_DEBUG("acceptor accept error: %s",e.message().c_str());
        }
        return;
    }

    template <typename CO>
    void accept(socket_type& sock,CO co) {
        BOOST_ASSERT(co != NULL);
        boost::system::error_code e;
        accept(sock,co,e);
        ////////////////////////////////////
        if(e) {
            throw_error(e,"acceptor accept");
        }
    }


};

}
}


#endif





<|start_filename|>unit_test/test_worker.cpp<|end_filename|>
// ====================================================================================
// Copyright (c) 2012, <EMAIL> All rights reserved.
// File         : test_worker.cpp
// Author       : <EMAIL>
// Last Change  : 11/19/2012 12:35 AM
// Description  : 
// ====================================================================================

#include <unistd.h>
#include <string>
#include <stdio.h>
#include <boost/thread.hpp>
#include <boost/bind.hpp>

#include "../orchid/all.hpp"


void test(orchid::coroutine_handle co) {
    int a = 1,b = 2;
    int c;
    orchid::run_in_thread([a,b,&c](){
        boost::this_thread::sleep(boost::posix_time::seconds(3));
        c = a+b;
    },co);

    ORCHID_DEBUG("c:%d",c);
}

int main(int argc,const char* argv[]) {
    orchid::scheduler sche;
    orchid::worker_pool pool(1);
    boost::thread t([&pool](){ pool.run(); });

    sche.spawn(test);
    sche.spawn([&pool](orchid::coroutine_handle co){
        int a = 1,b = 2;
        int c;
        pool.post([a,b,&c](){
            boost::this_thread::sleep(boost::posix_time::seconds(3));
            c = a+b;
        },co);
        ORCHID_DEBUG("c:%d",c);
    });

    sche.run();
}

 


