ManagerServant.py

Go to the documentation of this file.

#!/usr/bin/env python
# -*- coding: euc-jp -*-



import copy
import sys
import threading
import time
from omniORB import CORBA
import OpenRTM_aist
import RTC,RTM,RTM__POA
import SDOPackage


class ManagerServant(RTM__POA.Manager):
  """
  """

  
  def __init__(self):
    self._mgr    = OpenRTM_aist.Manager.instance()
    self._owner  = None
    self._rtcout = self._mgr.getLogbuf("ManagerServant")
    self._isMaster = False
    self._masters = []
    self._slaves = []
    self._masterMutex = threading.RLock()
    self._slaveMutex = threading.RLock()
    self._objref = None

    config = copy.deepcopy(self._mgr.getConfig())

    if OpenRTM_aist.toBool(config.getProperty("manager.is_master"), "YES", "NO", True):
      # this is master manager
      self._rtcout.RTC_TRACE("This manager is master.")

      if (not self.createINSManager()):
        self._rtcout.RTC_WARN("Manager CORBA servant creation failed.")
        return
        
      self._isMaster = True
      self._rtcout.RTC_TRACE("Manager CORBA servant was successfully created.")
      return
    else:
      # this is slave manager
      self._rtcout.RTC_TRACE("This manager is slave.")
      try:
        owner = self.findManager(config.getProperty("corba.master_manager"))
        if not owner:
          self._rtcout.RTC_INFO("Master manager not found")
          return

        if not self.createINSManager():
          self._rtcout.RTC_WARN("Manager CORBA servant creation failed.")
          return

        self.add_master_manager(owner)
        owner.add_slave_manager(self._objref)
        return
      except:
        self._rtcout.RTC_ERROR("Unknown exception cought.")
        self._rtcout.RTC_ERROR(OpenRTM_aist.Logger.print_exception())
        
        
    return


  
  def __del__(self):
    guard_master = OpenRTM_aist.ScopedLock(self._masterMutex)
    for i in range(len(self._masters)):
      try:
        if CORBA.is_nil(self._masters[i]):
          continue
        self._masters[i].remove_slave_manager(self._objref)
      except:
        self._masters[i] = RTM.Manager._nil
    self._masters = []

    guard_slave = OpenRTM_aist.ScopedLock(self._slaveMutex)
    for i in range(len(self._slaves)):
      try:
        if CORBA.is_nil(self._slaves[i]):
          continue
        self._slaves[i].remove_master_manager(self._objref)
      except:
        self._slaves[i] = RTM.Manager._nil
    self._slaves = []

    del guard_slave
    del guard_master
    return


  
  def load_module(self, pathname, initfunc):
    self._rtcout.RTC_TRACE("ManagerServant::load_module(%s, %s)", (pathname, initfunc))
    self._mgr.load(pathname, initfunc)
    return RTC.RTC_OK


  
  def unload_module(self, pathname):
    self._rtcout.RTC_TRACE("ManagerServant::unload_module(%s)", pathname)
    self._mgr.unload(pathname)
    return RTC.RTC_OK
  

  
  def get_loadable_modules(self):
    self._rtcout.RTC_TRACE("get_loadable_modules()")

    # copy local module profiles
    prof = self._mgr.getLoadableModules()
    cprof = [ RTM.ModuleProfile([]) for i in prof ]

    for i in range(len(prof)):
      OpenRTM_aist.NVUtil.copyFromProperties(cprof[i].properties, prof[i])

    return cprof


  
  def get_loaded_modules(self):
    self._rtcout.RTC_TRACE("get_loaded_modules()")
    prof = self._mgr.getLoadedModules()
    cprof = [RTM.ModuleProfile([]) for i in prof]
    
    for i in range(len(prof)):
      OpenRTM_aist.NVUtil.copyFromProperties(cprof[i].properties, prof[i])

    return cprof


  
  def get_factory_profiles(self):
    self._rtcout.RTC_TRACE("get_factory_profiles()")
    prof = self._mgr.getFactoryProfiles()
    cprof = [RTM.ModuleProfile([]) for i in prof]
    
    for i in range(len(prof)):
      OpenRTM_aist.NVUtil.copyFromProperties(cprof[i].properties, prof[i])

    return cprof


  
  def create_component(self, module_name):
    self._rtcout.RTC_TRACE("create_component(%s)", module_name)

    arg = module_name
    pos0 = arg.find("&manager=")
    pos1 = arg.find("?manager=")

    if pos0 == -1 and pos1 == -1:
      # create on this manager
      rtc = self._mgr.createComponent(module_name)
      if not rtc:
        return RTC.RTObject._nil
      return rtc.getObjRef()

    # create other manager

    # extract manager's location
    # since Python2.5 
    # pos = (lambda x: pos0 if x == -1 else pos1)(pos0)
    if pos0 == -1:
      pos = pos1
    else:
      pos = pos0
    
    endpos = arg.find('&', pos + 1)
    if endpos == -1:
      mgrstr = arg[(pos + 1):]
    else:
      mgrstr = arg[(pos + 1): endpos]
    self._rtcout.RTC_VERBOSE("Manager arg: %s", mgrstr)
    mgrvstr = mgrstr.split(":")
    if len(mgrvstr) != 2:
      self._rtcout.RTC_WARN("Invalid manager name: %s", mgrstr)
      return RTC.RTObject._nil

    eqpos = mgrstr.find("=")
    if eqpos == -1:
      self._rtcout.RTC_WARN("Invalid argument: %s", module_name)
      return RTC.RTObject._nil

    mgrstr = mgrstr[eqpos + 1:]
    self._rtcout.RTC_DEBUG("Manager is %s", mgrstr)

    # find manager
    mgrobj = self.findManager(mgrstr)
    if CORBA.is_nil(mgrobj):
      cmd = "rtcd_python -p "
      cmd += mgrvstr[1] # port number

      self._rtcout.RTC_DEBUG("Invoking command: %s.", cmd)
      ret = OpenRTM_aist.launch_shell(cmd)
      if ret == -1:
        self._rtcout.RTC_DEBUG("%s: failed", cmd)
        return RTC.RTObject._nil

      # find manager
      time.sleep(0.01)
      count = 0
      while CORBA.is_nil(mgrobj):
        mgrobj = self.findManager(mgrstr)
        count += 1
        if count > 1000:
          break
        time.sleep(0.01)

    if CORBA.is_nil(mgrobj):
      self._rtcout.RTC_WARN("Manager cannot be found.")
      return RTC.RTObject._nil
    
    # create component on the manager
    if endpos == -1:
      arg = arg[:pos]
    else:
      arg = arg[:pos] + arg[endpos:]
    self._rtcout.RTC_DEBUG("Creating component on %s",  mgrstr)
    self._rtcout.RTC_DEBUG("arg: %s", arg)
    try:
      rtobj = mgrobj.create_component(arg)
      self._rtcout.RTC_DEBUG("Component created %s",  arg)
      return rtobj
    except CORBA.SystemException:
      self._rtcout.RTC_DEBUG("Exception was caught while creating component.")
      self._rtcout.RTC_ERROR(OpenRTM_aist.Logger.print_exception())
      return RTC.RTObject._nil
    except:
      self._rtcout.RTC_DEBUG(OpenRTM_aist.Logger.print_exception())

    return RTC.RTObject._nil

  
  
  def delete_component(self, instance_name):
    self._rtcout.RTC_TRACE("delete_component(%s)", instance_name)
    self._mgr.deleteComponent(instance_name)
    return RTC.RTC_OK
  

  
  def get_components(self):
    self._rtcout.RTC_TRACE("get_components()")

    # get local component references
    rtcs = self._mgr.getComponents()
    crtcs = []

    for rtc in rtcs:
      crtcs.append(rtc.getObjRef())

    # get slaves' component references
    self._rtcout.RTC_DEBUG("%d slave managers exists.", len(self._slaves))
    for i in range(len(self._slaves)):
      try:
        if not CORBA.is_nil(self._slaves[i]):
          srtcs = self._slaves[i].get_components()
          OpenRTM_aist.CORBA_SeqUtil.push_back_list(crtcs, srtcs)
          continue
      except:
        self._RTC_INFO("slave (%d) has disappeared.", i)
        self._slaves[i] = RTM.Manager._nil

      OpenRTM_aist.CORBA_SeqUtil.erase(self._slaves, i)
      i -= 1

    return crtcs
  

  
  def get_component_profiles(self):
    rtcs = self._mgr.getComponents()
    cprofs = [rtc.get_component_profile() for rtc in rtcs]

    # copy slaves' component profiles
    guard = OpenRTM_aist.ScopedLock(self._slaveMutex)
    self._rtcout.RTC_DEBUG("%d slave managers exists.", len(self._slaves))

    for i in range(len(self._slaves)):
      try:
        if not CORBA.is_nil(self._slaves[i]):
          sprofs = self._slaves[i].get_component_profiles()
          OpenRTM_aist.CORBA_SeqUtil.push_back_list(cprofs, sprofs)
          continue
      except:
        self._rtcout.RTC_INFO("slave (%d) has disappeared.", i)
        self._slaves[i] = RTM.Manager._nil

      OpenRTM_aist.CORBA_SeqUtil.erase(self._slaves, i)
      i -= 1

    del guard
    return cprofs


  
  def get_profile(self):
    self._rtcout.RTC_TRACE("get_profile()")
    prof = RTM.ModuleProfile([])
    OpenRTM_aist.NVUtil.copyFromProperties(prof.properties, self._mgr.getConfig().getNode("manager"))

    return prof
  

  
  def get_configuration(self):
    self._rtcout.RTC_TRACE("get_configuration()")
    nvlist = []
    OpenRTM_aist.NVUtil.copyFromProperties(nvlist, self._mgr.getConfig())
    return nvlist
  

  
  def set_configuration(self, name, value):
    self._rtcout.RTC_TRACE("set_configuration(name = %s, value = %s)", (name, value))
    self._mgr.getConfig().setProperty(name, value)
    return RTC.RTC_OK
  


  
  def is_master(self):
    # since Python2.5
    # self._rtcout.RTC_TRACE("is_master(): %s", (lambda x: "YES" if x else "NO")(self._isMaster))
    ret = ""
    if self._isMaster:
      ret = "YES"
    else:
      ret = "NO"
    self._rtcout.RTC_TRACE("is_master(): %s", ret)
    return self._isMaster


  
  def get_master_managers(self):
    self._rtcout.RTC_TRACE("get_master_managers()")
    guard = OpenRTM_aist.ScopedLock(self._masterMutex)
    
    return self._masters


  
  def add_master_manager(self, mgr):
    guard = OpenRTM_aist.ScopedLock(self._masterMutex)
    self._rtcout.RTC_TRACE("add_master_manager(), %d masters", len(self._masters))
    index = OpenRTM_aist.CORBA_SeqUtil.find(self._masters, self.is_equiv(mgr))
    
    if not (index < 0): # found in my list
      self._rtcout.RTC_ERROR("Already exists.")
      return RTC.BAD_PARAMETER
    
    OpenRTM_aist.CORBA_SeqUtil.push_back(self._masters, mgr)
    self._rtcout.RTC_TRACE("add_master_manager() done, %d masters", len(self._masters))
    del guard
    return RTC.RTC_OK

  
  
  def remove_master_manager(self, mgr):
    guard = OpenRTM_aist.ScopedLock(self._masterMutex)
    self._rtcout.RTC_TRACE("remove_master_manager(), %d masters", len(self._masters))

    index = OpenRTM_aist.CORBA_SeqUtil.find(self._masters, self.is_equiv(mgr))
    
    if index < 0: # not found in my list
      self._rtcout.RTC_ERROR("Not found.")
      return RTC.BAD_PARAMETER
    
    OpenRTM_aist.CORBA_SeqUtil.erase(self._masters, index)
    self._rtcout.RTC_TRACE("remove_master_manager() done, %d masters", len(self._masters))
    del guard
    return RTC.RTC_OK


  
  def get_slave_managers(self):
    guard = OpenRTM_aist.ScopedLock(self._slaveMutex)
    self._rtcout.RTC_TRACE("get_slave_managers(), %d slaves", len(self._slaves))
    return self._slaves


  
  def add_slave_manager(self, mgr):
    guard = OpenRTM_aist.ScopedLock(self._slaveMutex)
    self._rtcout.RTC_TRACE("add_slave_manager(), %d slaves", len(self._slaves))
    
    index = OpenRTM_aist.CORBA_SeqUtil.find(self._slaves, self.is_equiv(mgr))
    
    if not (index < 0): # found in my list
      self._rtcout.RTC_ERROR("Already exists.")
      return RTC.BAD_PARAMETER
    
    OpenRTM_aist.CORBA_SeqUtil.push_back(self._slaves, mgr)
    self._rtcout.RTC_TRACE("add_slave_manager() done, %d slaves", len(self._slaves))
    del guard
    return RTC.RTC_OK


  
  def remove_slave_manager(self, mgr):
    guard = OpenRTM_aist.ScopedLock(self._slaveMutex)
    self._rtcout.RTC_TRACE("remove_slave_manager(), %d slaves", len(self._slaves))
    index = OpenRTM_aist.CORBA_SeqUtil.find(self._slaves, self.is_equiv(mgr))
    
    if index < 0: # not found in my list
      self._rtcout.RTC_ERROR("Not found.")
      return RTC.BAD_PARAMETER
    
    OpenRTM_aist.CORBA_SeqUtil.erase(self._slaves, index)
    self._rtcout.RTC_TRACE("remove_slave_manager() done, %d slaves", len(self._slaves))
    del guard
    return RTC.RTC_OK


  
  def fork(self):
    # self._mgr.fork()
    return RTC.RTC_OK

  
  
  def shutdown(self):
    self._mgr.terminate()
    return RTC.RTC_OK

  
  
  def restart(self):
    # self._mgr.restart()
    return RTC.RTC_OK
  

  
  def get_service(self, name):
    return CORBA.Object._nil

  
  
  def getObjRef(self):
    return self._objref


  
  def createINSManager(self):
    try:
      poa = self._mgr.getORB().resolve_initial_references("omniINSPOA")
      poa._get_the_POAManager().activate()
      id = self._mgr.getConfig().getProperty("manager.name")
      poa.activate_object_with_id(id, self)
      mgrobj = poa.id_to_reference(id)
      self._objref = mgrobj._narrow(RTM.Manager)
    except:
      self._rtcout.RTC_DEBUG(OpenRTM_aist.Logger.print_exception())
      return False

    return True


  
  def findManager(self, host_port):
    self._rtcout.RTC_TRACE("findManager(host_port = %s)", host_port)
    try:
      config = copy.deepcopy(self._mgr.getConfig())
      mgrloc = "corbaloc:iiop:"
      mgrloc += host_port
      mgrloc += "/" + config.getProperty("manager.name")
      self._rtcout.RTC_DEBUG("corbaloc: %s", mgrloc)

      mobj = self._mgr.getORB().string_to_object(mgrloc)
      mgr = mobj._narrow(RTM.Manager)
      return mgr

    except CORBA.SystemException:
      self._rtcout.RTC_DEBUG(OpenRTM_aist.Logger.print_exception())
      
    except:
      self._rtcout.RTC_ERROR("Unknown exception cought.")
      self._rtcout.RTC_DEBUG(OpenRTM_aist.Logger.print_exception())

    return RTM.Manager._nil


  class is_equiv:
    def __init__(self, mgr):
      self._mgr = mgr

    def __call__(self, mgr):
      if not self._mgr or not mgr:
        return self._mgr == mgr

      return self._mgr._is_equivalent(mgr)