L:/channel/cvm/include/CvmBaseTask.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 _CVM_TASK_H_
#define _CVM_TASK_H_

#include "ace/Message_Block.h"
#include "ace/Task.h"
#include "ace/Thread_Mutex.h"
#include "ace/Thread_Semaphore.h"
#include "ace/Get_Opt.h"
#include "ace/Reactor.h"
#include "ace/Service_Object.h"
#include "ace/OS.h"
#include "ace/Signal.h"
#include "ace/Log_Msg.h"
#include "ace/Service_Repository.h"
#include "ace/Service_Types.h"
#include "ace/Dynamic_Service.h"

#include <Channel.h>
#include <CvmBaseChannel.h>

#include <string>
#include <iostream>


namespace cvm {

  typedef enum {
    TASK_NULL=0,
    TASK_INIT,
    TASK_ACTIVE,
    TASK_SUSPEND,
    TASK_EXIT
  } TASK_STATE;

  template <class Channel>
  class CvmBaseTask : public ACE_Task<ACE_MT_SYNCH> {
  public:
    typedef typename Channel::Port Port;
    
  protected:
    std::string chan_name_;
    int num_thr;

    Channel *my_chan_;
    Port *my_port_;

    std::string my_name_;
    TASK_STATE my_state_;
    ACE_Thread_Semaphore exit_sema_; //for thread group exit gracely
    ACE_Thread_Mutex exit_lock_; //for thread group exit gracely

  public:
    enum { QUEUE_MAX = 16 * 1024 };

    CvmBaseTask() : my_state_(TASK_NULL), exit_sema_(0) {
      num_thr = 1;
      my_chan_ = NULL;
      my_port_ = NULL;
    }

    TASK_STATE state(void) { return my_state_;}
    std::string my_name(void) { return my_name_; }
    std::string chan_name(void) { return chan_name_;}
    Channel * my_chan(void) { return my_chan_; }
    Port * my_port(void) { return my_port_;}
 
    virtual channel::Status prepare(void) //initialization before thread starts
      {
        //subscribe msg
        return channel::SUCCESS;
      }
    virtual channel::Status cleanup(void) //cleanup before threads exit
      {
        //unsubscribe msg
        return channel::SUCCESS;
      }
    virtual int work (void)      //main processing loop
      {
        ACE_DEBUG ((LM_DEBUG,
                    "(%t) %s dmb_task coming up ...\n", my_name().c_str()));

        return 0;
      }

    virtual int init (int argc, char *argv[])
      {
        ACE_TCHAR cn[128];
        ACE_TCHAR bn[128];
        my_state_ = TASK_INIT;
        cn[0]='\0';
        bn[0]='\0';

        ACE_Get_Opt     get_opt (argc, argv, "t:c:", 0);

        for (int c; (c = get_opt ()) != -1; )
          switch (c)
            {
            case 't':
              num_thr = ACE_static_cast
                (int, ACE_OS::atoi (get_opt.opt_arg ()));        
              break;
            case 'c': // channel name
              ACE_OS::strsncpy
                (cn, get_opt.opt_arg (), 128);
              break;
            default:
              break;
            }

        chan_name_ = cn;
        if (chan_name_.length() == 0) {
          ACE_DEBUG ((LM_DEBUG,
                      "(%t) Task not bound to any Channel...\n"));
          return -1;
        }
  
        if (chan_name_.length() > 0) {
          my_chan_ = CvmBaseChannel<Channel>::find_chan(cn);
          if (my_chan_ == NULL) {
            ACE_DEBUG ((LM_DEBUG,
                        "(%t) fail to find channel[%s]...\n", cn));
            return -1;
          }

          my_port_ = new Port(my_chan_,msg_queue());
        }
        return open(NULL);
      }
    virtual int fini ()
      {
        ACE_DEBUG((LM_DEBUG, "..fini 1\n"));

        my_state_ = TASK_EXIT;
        ACE_DEBUG((LM_DEBUG, "(%t) Task [%s] fini...\n", my_name().c_str()));
        if(msg_queue()->state() == ACE_Message_Queue_Base::PULSED) {
          ACE_DEBUG ((LM_DEBUG, "task %s has been suspendded, please resume it before remove it\n", my_name().c_str() ));
          return -1;
        }

        ACE_DEBUG((LM_DEBUG, "..fini 2\n"));

        if(msg_queue()->state() != ACE_Message_Queue_Base::DEACTIVATED) {
          cleanup();
          msg_queue()->close();
        }
        ACE_DEBUG((LM_DEBUG, "..fini 3\n"));

        //since AppTask threads are detached, we cannot wait for them
        //use semaphore to synchronise the process
        for(int i=0; i<num_thr; i++)
          exit_sema_.acquire();

        //we still need wait some time for OS cleanup Task threads
        ACE_OS::sleep(1);

        //clean up port
        delete my_port_;

        //ACE_DEBUG((LM_DEBUG, "..fini 4\n"));

        return 0;
      }
    virtual int open (void *)
      {
        if(prepare() == channel::FAILURE) {
          ACE_DEBUG ((LM_DEBUG,
                      "(%t) fail inside prepare()...\n"));
          return -1;
        }

        ACE_DEBUG ((LM_DEBUG,
                    "(%t) finish prepare()...\n"));
        if (msg_queue ()->activate ()
            == ACE_Message_Queue_Base::ACTIVATED) {
          ACE_DEBUG ((LM_DEBUG,
                      "(%t) finish activate_que()...\n"));
          msg_queue ()->high_water_mark (QUEUE_MAX);

          ACE_DEBUG ((LM_DEBUG, "%d app threads start...\n", num_thr));

          my_state_ = TASK_ACTIVE;
          return activate (THR_NEW_LWP | THR_DETACHED, num_thr);
        } else {
          ACE_DEBUG ((LM_DEBUG,
                      "(%t) fail to activate msg_que...\n"));
          return -1;
        }
      }
    virtual int close (u_long)
      {
        my_state_ = TASK_EXIT;
        //should implement reference counting, so we only
        //cleanup when the last thread exits
        ACE_DEBUG ((LM_DEBUG,  "Task [%s] close...\n", my_name().c_str()));
        ACE_GUARD_RETURN(ACE_Thread_Mutex, guard, exit_lock_, 0);
        if(msg_queue()->state() != ACE_Message_Queue_Base::DEACTIVATED) {
          cleanup();
          msg_queue()->close();
        }
        exit_sema_.release();

        return 0;
      }
    virtual int put (ACE_Message_Block *mb, ACE_Time_Value *t)
      {
        return msg_queue ()->enqueue_tail
          (mb, (ACE_Time_Value *) &t);
      }
    virtual int info (char **strp, size_t length) const
      {
        char buf[256];

        ACE_OS::sprintf (buf,
                         "%s dmb_task coming up ... \n",
                         my_name_.c_str());

        if (*strp == 0 && (*strp = ACE_OS::strdup (buf)) == 0)
          return -1;
        else
          ACE_OS::strncpy (*strp, buf, length);
        return ACE_OS::strlen (buf);
      }
    virtual int suspend()
      {
        my_state_ = TASK_SUSPEND;
        my_port_->suspend();
        return 0; //add later
      }
    virtual int resume()
      {
        if(my_state_ == TASK_SUSPEND) {
          my_state_ = TASK_ACTIVE;
          my_port_->resume();
        }
        return 0; //add later
      }
    virtual int svc (void)   
      {
        my_name_ = rtrv_task_name();

        ACE_DEBUG ((LM_DEBUG,
                    "(%t) %s starts running ...\n", my_name().c_str()));

        return work();
      }     

  protected:

    const ACE_TCHAR *rtrv_task_name (void)
      {
        ACE_Service_Repository_Iterator iter
          (*ACE_Service_Repository::instance (), 0);

        for (const ACE_Service_Type *st;
             iter.next (st) != 0; iter.advance()) {
          const ACE_Service_Type_Impl *type = st->type();
          if (type == 0) continue;
          if (((ACE_Service_Object *)this) == type->object ())
            return st->name ();
        }
  
        return "Unamed Task";
      }

  };

};


#endif

---