ACE(Adaptive Communication Environment)

다중 이벤트를 처리하기 위한 전통적인 방법 // 서버가 동시에 여러 네트워크 연결을 처리
    새로운 프로세스 또는 쓰레드를 생성하여 각각의 이벤트를 처리하는 방식

디멀티플렉서(demultiplexer)에 기초한 Reactor 모델 // Event-driven
    select(), poll(), WaitForMultipleObjects()

    개발자의 비지니스 로직 부분을 이벤트 핸들링과 분리
        Reactor와 Event Handler

Proactor pattern
    Demultiplexing이 socket event에 의해서 수행되는 점은 reactor pattern과 동일하다.
    Demultiplexing과정을 통해 분리된 메시지가 저장되는 Message queue를 가지고 있다.
    Message queue를 감시하는 스레드 풀을 가지고 있다.
    Message queue에 메시지가 저장되는 즉시 감시중이던 스레드가 메시지를 뽑아내어 적절한 메시지 핸들러를 호출하여 처리한다.
    Demultiplexing과 Message Handling 을 각각 다른 스레드풀에게 담당하도록 할 수 있으므로, 단순한 Reactor pattern보다 서버의 반응속도가 빨라진다.


ACE_Svc_Handler -> ACE_Task
    .open() // virtual
    .handle_close() // virtual
    .handle_input() // virtual
    .handle_output() // virtual
   
ACE_Acceptor -> ACE_Service_Object
    .open()

ACE_Reactor
    .instance() // static
        .run_reactor_event_loop()

#include <ace/OS.h>
#include <ace/Log_Msg.h>
#include <ace/Message_Block.h>
#include <ace/INET_Addr.h>
#include <ace/Svc_Handler.h>
#include <ace/SOCK_Acceptor.h>
#include <ace/SOCK_Stream.h>
#include <ace/Synch_Traits.h>
#include <ace/Reactor.h>
#include <ace/Acceptor.h>
#include <ace/Reactor_Notification_Strategy.h>

class Stream_Handler : public ACE_Svc_Handler<ACE_SOCK_STREAM, ACE_NULL_SYNCH> {
private :
    typedef ACE_Svc_Handler<ACE_SOCK_STREAM, ACE_NULL_SYNCH> super;
    ACE_INET_Addr remote_addr_;
    ACE_Reactor_Notification_Strategy noti_;

public :
    Stream_Handler() : noti_(0, this, ACE_Event_Handler::WRITE_MASK) { /* empty */ }

    //override
    virtual int open(void * = 0) {
        ACE_TRACE("Stream_Handler::open");

        if( super::open() == -1 ) return -1;

        noti_.reactor(this->reactor());
        this->msg_queue()->notification_strategy(&noti_);
        if( this->peer().get_remote_addr(remote_addr_) == 0 ) {
            ACE_DEBUG((LM_INFO, "[DEBUG%T](%N:%l) ### New client accepted: %s:%u\n",
                remote_addr_.get_host_addr(), remote_addr_.get_port_number()));
        }

        return 0;
    }

    //override
    virtual int handle_close(ACE_HANDLE handle, ACE_Reactor_Mask close_mask) {
        ACE_TRACE("Stream_Handler::handle_close");

        ACE_DEBUG((LM_INFO, "[DEBUG%T](%N:%l) ### Connection close %s:%u\n",
                remote_addr_.get_host_addr(), remote_addr_.get_port_number()));
        return super::handle_close(handle, close_mask);
    }

    //override
    virtual int handle_input(ACE_HANDLE handle = ACE_INVALID_HANDLE) {
        ACE_TRACE("Stream_Handler::override");

        char buf[1024];
        ssize_t recv_cnt;

        if( (recv_cnt = this->peer().recv(buf, 1024)) <= 0 ) return -1;

        ACE_Message_Block *mb;

        ACE_NEW_RETURN(mb, ACE_Message_Block(buf, recv_cnt), -1);
        mb->wr_ptr(recv_cnt);
        this->putq(mb);
        return 0;
    }

    //override
    virtual int handle_output(ACE_HANDLE handle = ACE_INVALID_HANDLE) {
        ACE_TRACE("Stream_Handler::handle_output");

        ACE_Message_Block *mb;

        ACE_Time_Value nowait(ACE_OS::gettimeofday());
        while ( this->getq(mb, &nowait) != -1 ) {
            ssize_t send_cnt = this->peer().send(mb->rd_ptr(), mb->length());
            if( send_cnt == -1 ) {
                ACE_ERROR((LM_ERROR, "[ERROR%T](%N:%l) ### %p\n",
                                "Stream_Handler::handle_output"));
            } else {
                mb->rd_ptr(send_cnt);
            }

            if( mb->length() > 0 ) {
                this->ungetq(mb);
                break;
            }

            mb->release();
        }

        if( this->msg_queue()->is_empty() )
            this->reactor()->cancel_wakeup(this, ACE_Event_Handler::WRITE_MASK);
        else
            this->reactor()->schedule_wakeup(this, ACE_Event_Handler::WRITE_MASK);

        return 0;
    }   
};

int ACE_TMAIN(int argc, ACE_TCHAR *argv[])
{
    ACE_Acceptor<Stream_Handler, ACE_SOCK_ACCEPTOR> acceptor;
    ACE_INET_Addr listen;

    listen.set(9088);
    acceptor.open(listen);
    ACE_Reactor::instance()->run_reactor_event_loop();
    ACE_RETURN(0);
}

ACE_Service_Handler -> ACE_Handler
    .open() // virtual
    .addresses()
    .handle_read_stream()
    .handle_write_stream()

ACE_Asynch_Acceptor -> ACE_Handler
    .open()

ACE_Proactor
    .instance() // static
        proactor_run_event_loop()

#include <ace/OS.h>
#include <ace/Log_Msg.h>
#include <ace/Message_Block.h>
#include <ace/INET_Addr.h>
#include <ace/Proactor.h>
#include <ace/Asynch_IO.h>
#include <ace/Asynch_Acceptor.h>

class Stream_Handler : public ACE_Service_Handler {
private :
    ACE_Asynch_Read_Stream    reader_;
    ACE_Asynch_Write_Stream writer_;
    ACE_INET_Addr            remote_address_;

public :
    ~Stream_Handler() {
        if( this->handle() != ACE_INVALID_HANDLE )
            ACE_OS::closesocket(this->handle());
    }

    //override
    virtual void open(ACE_HANDLE new_handle, ACE_Message_Block &message) {
        ACE_TRACE("Stream_Handler::open");

        ACE_ASSERT(new_handle != ACE_INVALID_HANDLE);
        this->handle(new_handle);
        if( reader_.open(*this) != 0 || writer_.open(*this) != 0 ) {
            ACE_ERROR((LM_ERROR, "[ERROR%T](%N:%l) ### %p\n", "Stream_Handler::open"));
            delete this;
            return;
        }

        ACE_Message_Block *mb;

        ACE_NEW_NORETURN(mb, ACE_Message_Block(1024));
        if( reader_.read(*mb, mb->space()) != 0 ) {
            ACE_ERROR((LM_ERROR, "[ERROR%T](%N:%l) ### %p\n", "Stream_Handler::open"));
            delete this;
            return;
        }

        ACE_DEBUG((LM_INFO, "[DEBUG%T](%N:%l) ### New client accepted: %s:%u\n",
                remote_address_.get_host_addr(), remote_address_.get_port_number()));
    }

    //override
    virtual void addresses(const ACE_INET_Addr &remote_address, const ACE_INET_Addr &local_address) {
        ACE_TRACE("Stream_Handler::addresses");

        ACE_UNUSED_ARG(local_address);
        remote_address_ = remote_address;
    }
   
    //override
    virtual void handle_read_stream(const ACE_Asynch_Read_Stream::Result &result) {
        ACE_TRACE("Stream_Handler::handle_read_stream");

        ACE_Message_Block &mb = result.message_block();
        if( !result.success() || result.bytes_transferred() == 0 ) {
            ACE_DEBUG((LM_INFO, "[DEBUG%T](%N:%l) ### Connection close %s:%u\n",
                remote_address_.get_host_addr(), remote_address_.get_port_number()));
            mb.release();
            delete this;
        } else {
            if ( writer_.write (mb, mb.length ()) != 0 ) {
                ACE_ERROR((LM_ERROR, "[ERROR%T](%N:%l) ### %p\n",
                                       "Stream_Handler::handle_read_stream"));
                mb.release();
                delete this;
            } else {
                ACE_Message_Block *new_mb;

                ACE_NEW_NORETURN(new_mb, ACE_Message_Block(1024));
                reader_.read(*new_mb, new_mb->space());
            }
        }
    }
   
    //override
    virtual void handle_write_stream(const ACE_Asynch_Write_Stream::Result &result) {
        ACE_TRACE("Stream_Handler::handle_write_stream");

        result.message_block().release();
    }
};

int ACE_TMAIN(int agrc, ACE_TCHAR *argv[])
{
    ACE_Asynch_Acceptor<Stream_Handler> acceptor;

    ACE_INET_Addr listen;

    listen.set(9088);
    acceptor.open(listen,
        0,    // byte to read
        1,    // call pass_address
        ACE_DEFAULT_ASYNCH_BACKLOG,
        1,    // reuse_addr
        0,    // default proactor
        0,    // validate new connection
        1,    // reissue accept
        -1);// number of initial accepts
    ACE_Proactor::instance()->proactor_run_event_loop();
    ACE_RETURN(0);
}

참조 사이트:

http://lyb1495.tistory.com/entry/ACE-Reactor%EA%B8%B0%EB%B0%98-echo-server

http://lyb1495.tistory.com/entry/ACE-Proactor%EA%B8%B0%EB%B0%98-%EB%B9%84%EB%8F%99%EA%B8%B0asynchronous-echo-server
http://coms.tistory.com/17

http://lyb1495.tistory.com/entry/ACEADAPTIVE-Communication-Environment

http://www.redwiki.net/wiki/wiki.php/Proactor
http://www.redwiki.net/wiki/wiki.php/Overview%20of%20ACE
http://www.redwiki.net/wiki/wiki.php/ACE
http://www.cs.wustl.edu/~schmidt/ACE.html