L:/channel/channel/include/OutputMgr.h

Go to the documentation of this file.

// Copyright (c) 2005, 2006 Yigong Liu
// Permission to use, copy, modify, distribute and sell this software for any 
//     purpose is hereby granted without fee, provided that the above copyright 
//     notice appear in all copies and that both that copyright notice and this 
//     permission notice appear in supporting documentation.
// The author makes no representations about the 
//     suitability of this software for any purpose. It is provided "as is" 
//     without express or implied warranty.

#ifndef _OUTPUT_MGR_
#define _OUTPUT_MGR_

#include "ace/Message_Block.h"
#include "ace/Task.h"
#include "ace/Thread_Manager.h"
#include "ace/Thread_Mutex.h"
#include "ace/Thread_Semaphore.h"
#include "ace/Log_Msg.h"
#include "ace/Signal.h"
#include "ace/CDR_Stream.h"
//for ACE_GUARD macros, need the following 2 headers
#include "ace/Synch.h"
#include "ace/OS.h"

namespace channel {

  template <class Channel, class ConnHandler>
  class Out_Bound_Msg {
  public:
    typedef typename Channel::Msg Msg;
    Msg *msg;
    ConnHandler *conn;
    Out_Bound_Msg (Msg *m, ConnHandler *c) : msg(m), conn(c) {
    }
  };


  template <class Channel, class ConnHandler>
  class OutputMgr
    : public ACE_Task<ACE_MT_SYNCH> {
    private:
    int num_thr_;
    ACE_Thread_Mutex num_thr_lock_; //for gracely shutdown
    ACE_Thread_Semaphore exit_sema_; //for thread group exit gracely

    public:
    typedef typename Channel::Msg Msg;

    enum { QUEUE_MAX = 16 * 1024 }; //16k

    OutputMgr(int nt=1) {
      num_thr_ = nt;
    }

    int num_thr(void) { return num_thr_; }
  
    void num_thr(int nt) { num_thr_ = nt; }

    virtual int open () {// Initialization hook method.
      if (msg_queue ()->activate ()
          == ACE_Message_Queue_Base::ACTIVATED) {
        msg_queue ()->high_water_mark (QUEUE_MAX);
        return activate (THR_NEW_LWP | THR_DETACHED, num_thr_);
      } else return 0;
    }

    void shut_down(void) 
      {
        if(msg_queue()->state() != ACE_Message_Queue_Base::DEACTIVATED) {
          msg_queue()->deactivate(); //hope all blocked sending threads will wake up
        }
        for(int i=0; i<num_thr_; i++)
          exit_sema_.acquire();
      }

    virtual int close() { //is invoked in each thread
      //NO need to delete, since this is statically allocated by Connector
      //ACE_GUARD_RETURN(ACE_Thread_Mutex, guard, num_thr_lock_, -1);
      //num_thr_--;
      //if(num_thr_==0) //i am the last guy
      //  delete this;
      ACE_DEBUG((LM_DEBUG, "OutputMgr close down, thread exit...\n"));
      exit_sema_.release();
      return 0;
    }

    virtual int put (ACE_Message_Block *mb, ACE_Time_Value *timeout = 0)
      {
        return putq (mb, timeout);
      }

    protected:
    virtual int svc ()
      {
        ACE_Sig_Action no_sigpipe ((ACE_SignalHandler) SIG_IGN);
        ACE_Sig_Action original_action;
        no_sigpipe.register_action (SIGPIPE, &original_action);
 
        for (;;) {
          ACE_Message_Block *mblk = 0;
          if (getq (mblk) == -1) {
            if (errno == ESHUTDOWN) { //msgque deactivated
              break;
            } else if (errno != EWOULDBLOCK) break;
          } else {
            if (send (mblk) == -1) {
              ACE_DEBUG((LM_DEBUG, "(%t) failed to send one msg...\n"));
            }
            mblk->release();
          }
        }

        no_sigpipe.restore_action (SIGPIPE, original_action);
        return 0;
      }
    virtual int send (ACE_Message_Block *mblk)
      {
        Out_Bound_Msg<Channel, ConnHandler> *out_msg = (Out_Bound_Msg<Channel, ConnHandler> *)mblk->rd_ptr();
        Msg *msg = out_msg->msg;
        ConnHandler *connhndl = out_msg->conn;
        return connhndl->send_msg(msg);
      }
  };

};

#endif

---