home *** CD-ROM | disk | FTP | other *** search
/ OS/2 Shareware BBS: 10 Tools / 10-Tools.zip / fnb101.zip / Lib / site-packages / Fnorb / orb / GIOPClientWorkerThreaded.py < prev    next >
Text File  |  1999-06-28  |  11KB  |  393 lines
  1. #!/usr/bin/env python
  2. #############################################################################
  3. # Copyright (C) DSTC Pty Ltd (ACN 052 372 577) 1997, 1998, 1999
  4. # All Rights Reserved.
  5. #
  6. # The software contained on this media is the property of the DSTC Pty
  7. # Ltd.  Use of this software is strictly in accordance with the
  8. # license agreement in the accompanying LICENSE.HTML file.  If your
  9. # distribution of this software does not contain a LICENSE.HTML file
  10. # then you have no rights to use this software in any manner and
  11. # should contact DSTC at the address below to determine an appropriate
  12. # licensing arrangement.
  13. # 
  14. #      DSTC Pty Ltd
  15. #      Level 7, GP South
  16. #      Staff House Road
  17. #      University of Queensland
  18. #      St Lucia, 4072
  19. #      Australia
  20. #      Tel: +61 7 3365 4310
  21. #      Fax: +61 7 3365 4311
  22. #      Email: enquiries@dstc.edu.au
  23. # 
  24. # This software is being provided "AS IS" without warranty of any
  25. # kind.  In no event shall DSTC Pty Ltd be liable for damage of any
  26. # kind arising out of or in connection with the use or performance of
  27. # this software.
  28. #
  29. # Project:      Fnorb
  30. # File:         $Source: /units/arch/src/Fnorb/orb/RCS/GIOPClientWorkerThreaded.py,v $
  31. # Version:      @(#)$RCSfile: GIOPClientWorkerThreaded.py,v $ $Revision: 1.7 $
  32. #
  33. #############################################################################
  34. """ GIOPClientWorkerThreaded class. """
  35.  
  36.  
  37. # Standard/built-in modules.
  38. import thread
  39.  
  40. # Fnorb modules.
  41. import CORBA, GIOPClientWorker, GIOPConnectionHandler
  42.  
  43. # Fnorb threading modules.
  44. import condvar
  45.  
  46.  
  47. class GIOPClientWorkerThreaded(GIOPClientWorker.GIOPClientWorker):
  48.     """ GIOPClientWorkerThreaded class. """
  49.  
  50.     def __init__(self, protocol, address):
  51.     """ Provide an interface to a remote object!
  52.  
  53.     'protocol' is the protocol used by the client.
  54.     'address'  is the address of the remote object.
  55.  
  56.     """
  57.     # fixme: We don't need to keep these as instance variables... (maybe
  58.     # for unpickling?),
  59.     self.__protocol = protocol
  60.     self.__address = address
  61.  
  62.     # This flag is set when we get a CloseConnection message, or an
  63.     # exception event.
  64.     self.__closed = 0
  65.  
  66.     # Each request has a unique id. associated with it.
  67.     self.__request_id = 0
  68.  
  69.     # Mutex to make access to the request id and the closed flag
  70.     # thread-safe.
  71.     self.__lk = thread.allocate_lock()
  72.  
  73.     # Condition variable for outgoing messages.
  74.     self.__outgoing_cv = condvar.condvar()
  75.  
  76.     # Dictionary of condition variables for threads that are blocked
  77.     # waiting for replies.
  78.     self.__blocked_pending = {} # {RequestId: condvar}
  79.  
  80.     # Dictionary of complete replies.
  81.     self.__replies = {} # {RequestId: Message}
  82.  
  83.     # Mutex for incoming messages.
  84.     self.__incoming_lk = thread.allocate_lock()
  85.  
  86.     # Ask the protocol to create a connection and actually connect to the
  87.     # remote object.
  88.     self.__connection = protocol.create_connection()
  89.     self.__connection.connect(self.__address)
  90.  
  91.     # Set the connection to BLOCKING mode.
  92.     self.__connection.blocking(1)
  93.  
  94.     # Create a handler to look after the connection.
  95.     self.__handler = GIOPConnectionHandler.GIOPConnectionHandler \
  96.              (self, self.__connection)
  97.  
  98.     # Start the read thread.
  99.     thread.start_new_thread(self.__read_thread, ())
  100.  
  101.     return
  102.  
  103.     def pseudo__del__(self):
  104.     """ Pseudo destructor to remove circular references. """
  105.  
  106.     # Close down the worker.
  107.     self.__close()
  108.  
  109.     # The handler holds a (circular) reference to this instance so we
  110.     # have to explicitly clean it up.
  111.     self.__handler.pseudo__del__()
  112.  
  113.     # Clean up *our* reference to *it*!
  114.     del self.__handler
  115.  
  116.     return
  117.  
  118.     #########################################################################
  119.     # GIOPClientWorker interface.
  120.     #########################################################################
  121.  
  122.     def send(self, request_id, message):
  123.     """ Send an operation request to the remote object.
  124.  
  125.     This method BLOCKS until the message has been sent.
  126.  
  127.     """
  128.     # Block until the connection is available.
  129.     self.__outgoing_cv.acquire()
  130.     try:
  131.         # Send the entire message.
  132.         n = 0
  133.         while n < len(message):
  134.         n = n + self.__handler.send(message)
  135.  
  136.     except CORBA.COMM_FAILURE, ex:
  137.         # Let somebody else have a go ;^)
  138.         self.__outgoing_cv.release()
  139.  
  140.         # If the connection has been closed then we can try again.
  141.         if self.is_closed():
  142.         raise CORBA.TRANSIENT()
  143.  
  144.         else:
  145.         raise ex
  146.  
  147.     # Let somebody else have a go ;^)
  148.     self.__outgoing_cv.release()
  149.  
  150.     return
  151.  
  152.     def recv(self, request_id):
  153.     """ Wait for a specific reply. """
  154.  
  155.     # Make sure that we aren't interfering with the read thread.
  156.     self.__incoming_lk.acquire()
  157.  
  158.     # See if the reply has already arrived.
  159.     reply = self.__replies.get(request_id)
  160.     if reply is None:
  161.         # Create a new condition variable.
  162.         cv = condvar.condvar()
  163.  
  164.         # Add the condition variable to the dictionary of blocked threads.
  165.         # The read thread will 'signal' the condition variable (if and)
  166.         # when the corresponding reply is received.
  167.         self.__blocked_pending[request_id] = cv
  168.  
  169.         # Acquire the condition variable so that if the read thread
  170.         # receives the reply before we have called 'wait', it will be
  171.         # blocked.
  172.         cv.acquire()
  173.  
  174.         # Let the read thread do its thang!
  175.         self.__incoming_lk.release()
  176.  
  177.         # Wait on the condition variable ('wait' releases the lock on the
  178.         # condition variable first).
  179.         cv.wait()
  180.         cv.release()
  181.  
  182.         self.__incoming_lk.acquire()
  183.         reply = self.__replies[request_id]
  184.         del self.__replies[request_id]
  185.         self.__incoming_lk.release()
  186.  
  187.          # If the condition variable was signalled because we received a
  188.          # 'CloseConnection' message, then the reply will be a
  189.         # 'CORBA.TRANSIENT' exception. If the condition variable was
  190.         # signalled because of some other failure then the reply will be
  191.         # an instance of the appropriate CORBA system exception.
  192.         if isinstance(reply, CORBA.SystemException):
  193.         raise reply
  194.  
  195.     # The reply has already arrived - now that's what I call service ;^)
  196.     else:
  197.         del self.__replies[request_id]
  198.         self.__incoming_lk.release()
  199.  
  200.     return reply
  201.  
  202.     def poll(self, request_id):
  203.     """ Poll for a reply to a specific request. """
  204.  
  205.     # Have we got our reply yet
  206.     self.__incoming_lk.acquire()
  207.     result = self.__replies.has_key(request_id)
  208.     self.__incoming_lk.release()
  209.  
  210.     return result
  211.  
  212.     def peek(self, request_id):
  213.     """ Peek at the reply for the specified request.
  214.  
  215.     This method does *not* delete the reply from the client's queue.
  216.  
  217.     """
  218.     # Get the reply.
  219.     self.__incoming_lk.acquire()
  220.     reply = self.__replies[request_id]
  221.     self.__incoming_lk.release()
  222.  
  223.     return reply
  224.  
  225.     def delete_reply(self, request_id):
  226.     """ Delete the reply with the specified request id. """
  227.  
  228.     # Delete the reply.
  229.     self.__incoming_lk.acquire()
  230.     del self.__replies[request_id]
  231.     self.__incoming_lk.release()
  232.  
  233.     return
  234.  
  235.     def next_request_id(self):
  236.     """ Return the next request id. """
  237.  
  238.     # fixme: This will get stupidly big oneday!!!!
  239.     self.__lk.acquire()
  240.     request_id = self.__request_id = self.__request_id + 1
  241.     self.__lk.release()
  242.  
  243.     return request_id
  244.  
  245.     def is_closed(self):
  246.     """ Has the worker received a close event? """
  247.  
  248.     self.__lk.acquire()
  249.     closed = self.__closed
  250.     self.__lk.release()
  251.  
  252.     return closed
  253.  
  254.     def close_connection(self):
  255.     """ Close down the connection.
  256.  
  257.     This method is only ever called from the read thread.
  258.  
  259.     """
  260.     # Unblock all threads that are waiting for replies.
  261.     self.__unblock_all(CORBA.TRANSIENT())
  262.  
  263.     # Close down the worker.
  264.     self.__close()
  265.  
  266.     return
  267.  
  268.     #########################################################################
  269.     # GIOPConnectionHandlerListener interface.
  270.     #########################################################################
  271.  
  272.     def message_received(self, message):
  273.     """ Called when a complete GIOP message has been received. """
  274.  
  275.     # Code re-use! The protected method '_message_received' is implemented
  276.     # in the 'GIOPClientWorker' class.
  277.     #
  278.     # If the message is a GIOP Reply message, then this method will call
  279.     # our '_reply_received' method.
  280.     self._message_received(message)
  281.  
  282.     return
  283.  
  284.     def message_sent(self):
  285.      """ Called when a complete GIOP message has been sent.
  286.  
  287.     In the threaded worker, we block until the message has been sent
  288.     anyway, so we don't have anything to do here!
  289.  
  290.     """
  291.     pass
  292.  
  293.     #########################################################################
  294.     # Protected interface.
  295.     #########################################################################
  296.  
  297.     def _reply_received(self, reply_header, cursor):
  298.     """ Called by the '_message_received' method of GIOPClientWorker. """
  299.  
  300.     self.__incoming_lk.acquire()
  301.  
  302.     # Add the reply to our dictionary of complete replies.
  303.     self.__replies[reply_header.request_id] = (reply_header, cursor)
  304.  
  305.     # See if a thread is blocked waiting for this reply.
  306.     cv = self.__blocked_pending.get(reply_header.request_id)
  307.  
  308.     if cv is not None:
  309.         # Signal the blocked thread.
  310.         cv.acquire()
  311.         cv.signal()
  312.  
  313.         # Clean up.
  314.         del self.__blocked_pending[reply_header.request_id]
  315.  
  316.     self.__incoming_lk.release()
  317.     return
  318.  
  319.     #########################################################################
  320.     # Private interface.
  321.     #########################################################################
  322.  
  323.     def __read_thread(self):
  324.     """ Read messages from the server. """
  325.  
  326.     try:
  327.         while not self.is_closed():
  328.         # Blocking receive.
  329.         self.__handler.recv()
  330.  
  331.         # If we have received a 'CloseConnection' message, then we notify
  332.         # the GIOPClient by raising a 'TRANSIENT' system exception.
  333.         ex = CORBA.TRANSIENT()
  334.  
  335.     except CORBA.SystemException, ex:
  336.         # If we have received a 'CloseConnection' message, then we notify
  337.         # the GIOPClient by raising a 'TRANSIENT' system exception.
  338.         if self.is_closed():
  339.         ex = CORBA.TRANSIENT()
  340.  
  341.         # Otherwise, it was some other system exception, so close the 
  342.         # connection down.
  343.         else:
  344.         self.__close()
  345.  
  346.     # Unblock all threads that are waiting for replies.
  347.     self.__unblock_all(ex)
  348.  
  349.     # Explicitly exit the thread!
  350.     thread.exit()
  351.  
  352.     def __close(self):
  353.     """ Close down the worker. """
  354.  
  355.     self.__lk.acquire()
  356.     if not self.__closed:
  357.         # All done!
  358.         self.__closed = 1
  359.         self.__lk.release()
  360.  
  361.          # Close our connection.
  362.         self.__connection.disconnect()
  363.  
  364.     else:
  365.         self.__lk.release()
  366.  
  367.     return
  368.  
  369.     def __unblock_all(self, exception):
  370.     """ Unblock all waiting threads with the specified exception. """
  371.  
  372.     self.__incoming_lk.acquire()
  373.     # Set the reply for each blocked thread to be the specified exception.
  374.     for request_id in self.__blocked_pending.keys():
  375.         self.__replies[request_id] = exception
  376.  
  377.     # Get a list of condition variables to signal.
  378.     blocked_pending = self.__blocked_pending.values()
  379.  
  380.     # Clean them up!
  381.     self.__blocked_pending = {}
  382.  
  383.     self.__incoming_lk.release()
  384.  
  385.     # Unblock each thread.
  386.     for cv in blocked_pending:
  387.         cv.acquire()
  388.         cv.signal()
  389.  
  390.     return
  391.  
  392. #############################################################################
  393.