Open Mash Cross Reference
mash/tcl/degas/degasserver/alm-degas-server.tcl

   import TCP/Server/DegasControl
   import AnnounceListenManager
   import DegasAgent
   
   #------------------------------------------------------------------

   # Class AnnounceListenManager/DegasServer

   #

   # purpose :

   #   This is the heart and soul of the Degas server.  

  #   It processes the messages from other servers and other 

  #       clients.   It also sends "alive" message to other 

  #       server.

  #

  # members :

  #

  # programs_

  #   An array index by addr, port and key. The value of the

  #   array contains the value of key submitted by client at

  #   addr/port.

  #

  # alive_ 

  #   An array index by addr and port.  alive_($addr,$port) exists

  #   iff the server at addr:port is alive.  (The value does not

  #   really matters).

  #

  # alive_timeout_id_

  #   This is an array index by addr and port, the id of the 

  #   "after" timeout command.  

  #

  # alive_timeout_

  #   Timeout of an alive message, after which the server is

  #   consider dead.

  #

  # alive_interval_id_

  #   This is the "after" id of the after event for announcing

  #   my own alive message.

  #

  # alive_interval_

  #   This is the interval between announcement of alive message.

  #

  # num_of_server_

  #   Number of other servers on this control session.  This is 

  #   incremented everytime an "alive" message is received, and

  #   decremented everytime a "bye" is received, or timeout of

  #   an alive occurs.

  #

  # servable_

  #   Servable value for server A for client B will be stored in 

  #   servable_(A-addr, A-port, B-addr, B-port)

  #

  # ignore_servable_(addr,port)

  #   Servable message for client addr/port will be ignored this is

  #   not_defined.  This happens because my servable value is larger 

  #   than at least one of the other server, so it is not possible 

  #   for me to service the client.

  # 

  # self_servable_(addr,port)

  #   My servable value for client addr/port

  #

  # servable_count_(addr,port)

  #   How many servable value for client addr/port have been received.

  #

  # servable_id_

  #   This is an array index by addr and port, the id of the 

  #   "after" timeout command.  

  #

  # servable_timeout_

  #   Timeout of an servable message, after which the server with current

  #   minimum start providing service even if not all servable value has

  #   been received.

  # 

  # out_session_(addr,port)

  #   Output session for client addr/port.

  #

  # unique_port_

  #   A unique port number. It is initialized to $min_port_, and increments

  #   by 4 everytime, wrap around at $max_port_

  # 

  # port_in_use_(port)

  #   A array indexed by all ports in used.

  #

  # max_port_

  #   Maximum port number to use.

  #

  # min_port_

  #   Minimum port number to use.

  #

  # addr_in_use_(addr1,addr2)

  #   This is set iff multicast address 224.3.$addr1.$addr2 is 

  #   already in used.  This is to avoid collision between two

  #   randomly generated multicast address.

  #

  #------------------------------------------------------------------

  
  Class AnnounceListenManager/DegasServer -superclass {AnnounceListenManager}
  
  #------------------------------------------------------------------

  # AnnouceListenManager/DegasServer::init

  #

 # purpose : Constructor for a AnnouceListenManager/DegasServer obj

 # input   : spec - Specifies where the ALM should announce and listen.

 #                  Format is <ipaddr>/<port> or <port>.

 #           mtu  - Maximum size of a message.  A buffer of this size

 #                  will be allocated for storing message.

 # output  : none

 #------------------------------------------------------------------

 
 AnnounceListenManager/DegasServer public init {spec mtu} {
     $self instvar alive_timeout_
     $self instvar alive_interval_
     $self instvar servable_timeout_
     $self instvar num_of_server_
     $self instvar min_port_
     $self instvar max_port_
     $self instvar unique_port_
 
     set alive_timeout_ [$self get_option aliveTimeout]
     set servable_timeout_ [$self get_option servableTimeout]
     set alive_interval_ [$self get_option aliveInterval]
     set min_port_ [$self get_option minPort]
     set max_port_ [$self get_option maxPort]
     set unique_port_ $min_port_
     set num_of_server_ 0
 
     $self next $spec $mtu
 }
 
 
 #------------------------------------------------------------------

 # AnnouceListenManager/DegasServer::recv_announcement

 #

 # purpose : This is called whenever an announcement is received.

 #           Method handle_client_announcement and handle_server_announcemt

 #           is called to process the message respectively.

 # input   : addr - the ip address where this message is from.

 #           port - the port where this message is from

 #           m    - the announcement message itself, which should

 #                  have the following format.

 #           {

 #               DCP

 #               [ client | server ]

 #               <rest of messages>

 #           }

 #           l    - length of the message.

 # output  : none

 #------------------------------------------------------------------

 
 AnnounceListenManager/DegasServer private recv_announcement {addr port m l} {
     # Check magic word.

     set magic_word [lindex $m 0]
     if {$magic_word != "DCP"} {
         # This message is not for me.

         return
     }
 
     # Check message type (from server ? or client ?)

     set type [lindex $m 1]
     set m [lrange $m 2 end]
     if {$type == "server"} {
         $self handle_server_message $addr $port $m
     } elseif {$type == "client"} {
         $self handle_client_message $addr $port $m
     }
 
 }
 
 
 #------------------------------------------------------------------

 # AnnounceListenManager/DegasServer::handle_client_message

 #

 # purpose : The client annoucement is parsed, and the infomation

 #           is stored in an array indexed by address, port and 

 #           key in the message.  Then based on the action of the

 #           announcement, the following happens :

 #           - if a new program is submitted (action == new), then

 #             it calculate a "servable" value, and announce this

 #             value to all other servers the client.   

 #           - update and cancel are not implemented yet. 

 # input   : addr - the ip address where this message is from.

 #           port - the port where this message is from

 #           m    - the announcement message itself, which should

 #                  have the following format.

 #           {

 #               action               [new|update|cancel]

 #               input_session_spec   <addr>/<port>/<ttl>

 #               control_channel_port <port>

 #               sources              <list of sources>

 #               max_num_of_sources   <int> or "*"

 #               init_callback        <program>

 #               recv_frame_callback  <program>

 #               new_source_callback  <program>

 #               del_source_callback  <program>

 #               destroy_callback     <program>

 #               }

 # output  : none

 #------------------------------------------------------------------

 
 AnnounceListenManager/DegasServer private handle_client_message {addr port m} {
 
     # Determine action

     if {[lindex $m 0] != "action"} {
         puts stderr "error parsing announcement. expect \"action\"
             keyword but found [lindex $m 0]"
         return
 
     }
     set action [lindex $m 1]
     set $m [lrange $m 2 end]
 
     switch -exact $action {
         "request" {
                         puts "request received"
                 }
         "new" {
             # A new service request is received.  Calculate the

             # servable value for this client.  And announce this

             # value to all client.

 
             $self instvar programs_
             foreach {key value} $m {
                 set programs_($addr,$port,$key) $value
             }
 
             $self instvar self_servable_
             $self instvar ignore_servable_
             set servable [$self calc_servable $addr $port]
             set self_servable_($addr,$port) $servable
 
             # this is a new request, so we must pay attention

             # to this client again.

             if {[info exists ignore_servable_($addr,$port)]} {
                 unset ignore_servable_($addr,$port)
             }
 
             # announce my servable value to all other server.

             $self announce_servable $servable $addr $port
 
             # timeout - start service if still the minimum even if

             # not all servable value is received.

             $self instvar servable_id_
             $self instvar servable_timeout_
             set servable_id_ [after $servable_timeout_ "
                 if {![info exists ignore_servable_($addr,$port)]} {
                     $self begin_service $addr $port
             }" ]
         }
         "update" {
             puts stderr "update is not supported yet"
         }
         "cancel" {
             puts stderr "cancel is not supported yet"
         }
         default {
             puts stderr "error parsing announcement. expect
             \"new\", \"update\", or \"cancel\" but got $action"
             return
         }
     }
 }
 
 
 #------------------------------------------------------------------

 # AnnounceListenManager/DegasServer::handle_server_message

 #

 # purpose : This is called by recv_annoucnement to process

 #           a server message.

 # input   : addr - the server which sent the message.

 #           port - the port of the server which sent the message.

 #           m    - the message itself, with the following format :

 #           {

 #               [ alive |

 #                 bye | 

 #                 servable <servable value> <client_addr> <client_port> |

 #                 service <client_addr> <client_port> 

 #               ]

 #           }

 # output  : none

 #------------------------------------------------------------------

 
 AnnounceListenManager/DegasServer private handle_server_message {addr port m} {
     $self instvar alive_
     $self instvar num_of_server_
     $self instvar alive_timeout_id_
     $self instvar alive_timeout_
 
     set action [lindex $m 0]
     switch $action {
     "alive" {
         if {![info exists alive_($addr,$port)]} {
             # new server detected.

             incr num_of_server_ 
         }
         set alive_($addr,$port) 1
         # cancel the last timeout and add a new one.

         # if after alive_timeout_, this timeout is not cancel,

         # then the server is considered dead.               

         if {[info exists alive_timeout_id_]} {
             after cancel $alive_timeout_id_
         }
         set alive_timeout_id_ [after $alive_timeout_ "$self bye_bye_server $addr $port"]
 
     }
     "bye" {
         $self bye_bye_server $addr $port
     }
     "service" {
         puts stderr "service message is not processed currently"
     }
     "servable" {
         # We received a servable message from another 

         # server.  We check if we are the minimum.  If

         # we are not, we ignore further servable message

         # for this client.  If we are the minimum after

         # all messages are received, (we break tie based

         # on ip address.) we are the lucky one to provide

         # service to the client.  We announce the good new

         # to all other servers.

 
         $self instvar ignore_servable_
         $self instvar self_servable_
         $self instvar servable_count_
         $self instvar servable_
         $self instvar num_of_server_
         $self instvar servable_id_
 
         set servable [lindex $m 0]
         set client_addr [lindex $m 1]
         set client_port [lindex $m 2]
         if {$servable < $self_servable_($client_addr,$client_port)} {
             set ignore_servable_($client_addr,$client_port) 1
         }
         set servable_($client_addr,$client_port,$addr,$port) $servable
         incr servable_count_($client_addr,$client_port) 
         if {$servable_count_($client_addr,$client_port) >= $num_of_server_ 
             && ![info exists ignore_servable_($client_addr,$client_port)] } {
             after cancel servable_id_
             $self begin_service $client_addr $client_port
         }
     }
     default {
         puts "unknown token $action in server message"
     }
     }
 }
 
 
 #------------------------------------------------------------------

 # AnnouceListenManager/DegasServer::create_annoucement

 #

 # purpose : Create a new message.  This message can then be send

 #           by calling "announce".

 # input   : 

 # return  : A new message.

 #------------------------------------------------------------------

 
 AnnounceListenManager/DegasServer private create_announcement {} {
     return "Dummy Message"
 }
 
 
 
 #------------------------------------------------------------------

 # AnnounceListenManager/DegasServer::calc_servable

 #

 # purpose : Calculate a value (0 .. 1) to indicate if it is good

 #           to provide a service to this client.  1 is good and 

 #           0 is bad.

 # input   : addr, port - indicate which client are we trying to

 #                        serve.

 # return  : the servable value as float.

 #------------------------------------------------------------------

 
 AnnounceListenManager/DegasServer private calc_servable {addr port} {
     # Not implemented yet.

     return 1.0
 }
 
 
 #------------------------------------------------------------------

 # AnnounceListenManager/DegasServer::bye_bye_server

 #

 # purpose : A server is down.  Remove all records of that server,

 #           and possibly take over it's job, if any.

 # input   : addr,port - the address and port of the server we are

 #                       saying bye-bye to.

 # output  : none.

 #------------------------------------------------------------------

 
 AnnounceListenManager/DegasServer private bye_bye_server {addr port} {
     $self instvar alive_
     $self instvar num_of_server_
 
     unset alive_($addr,$port)
     incr num_of_server_ -1
 }
 
 
 #------------------------------------------------------------------

 # AnnounceListenManager/DegasServer::announce_servable

 #

 # purpose : Create a servable message and announce it to other

 #       servers.

 # input   : s - a servable value (float)

 # output  : none

 #------------------------------------------------------------------

 
 AnnounceListenManager/DegasServer private announce_servable {s addr port} {
     $self log "announcing servable $s $addr $port"
     $self announce "DCP\nserver\nservable $s $addr $port\n"
 }
 
 
 
 #------------------------------------------------------------------

 # AnnounceListenManager/DegasServer::annouce_alive

 #

 # purpose : Create an alive announcement and annount to other servers.

 # input   : none

 # output  : none

 #------------------------------------------------------------------

 
 AnnounceListenManager/DegasServer private announce_alive {} {
 #$self log "announcing alive"

     $self announce "DCP\nserver\nalive\n"
 }
 
 
 
 #------------------------------------------------------------------

 # AnnounceListenManager/DegasServer::announce_bye

 #

 # purpose : Create a bye announcement and announce to other servers.

 # input   : none

 # output  : none

 #------------------------------------------------------------------

 
 AnnounceListenManager/DegasServer private announce_bye {} {
     $self log "announcing bye"
     $self announce "DCP\nserver\nbye\n"
 }
 
 
 
 #------------------------------------------------------------------

 # AnnonceListenManager/DegasServer::announce_service

 #

 # purpose : Annouce to other servers that I am providing service to

 #           to some particular client.

 # input   : addr,port - the client I am going to service.

 #           session - output multicast session.

 #           control_port - port where client should connect for control

 #               channel.

 # output  : none.

 #------------------------------------------------------------------

 
 AnnounceListenManager/DegasServer private announce_service {addr port session control_port} {
     $self announce "DCP\nserver\nservice $addr $port $session $control_port"
 }
 
 
 #------------------------------------------------------------------

 # AnnouceListenManager/DegasServer::begin_service

 #

 # purpose : Start the service for a client

 # input   : addr,port - the client I am going to service.

 # output  : none

 # note    : this should be private, but put as public to make debugging

 #           simple.

 #------------------------------------------------------------------

 
 AnnounceListenManager/DegasServer public begin_service {addr port} {
     $self instvar programs_
     $self instvar output_session_
     $self instvar in_service_
 
     if {[info exists in_service_($addr,$port)]} {
         $self log "Duplicated service requested for $addr/$port"
         return
     }
     $self log "Starting service for $addr/$port"
     set in_service_($addr,$port) 1
 
     set output_session_($addr,$port) [$self generate_new_session_spec]
     set ctrl_port [$self generate_unique_port]
     puts "Output session : $output_session_($addr,$port)"
     set agent [new DegasAgent \
         [$self get_program $addr $port] $output_session_($addr,$port)]
     set ctrl_tcp_ [new TCP/Server/DegasControl $agent]
     $ctrl_tcp_ open $ctrl_port
     $self announce_service $addr $port $output_session_($addr,$port) $ctrl_port
 }
 
 
 
 #------------------------------------------------------------------

 # AnnouceListenManager/DegasServer::run

 #

 # purpose : Start the annouce listen manager.  This will cause the

 #           alive message to be multicast periodically.

 # input   : none

 # output  : none

 #------------------------------------------------------------------

 
 AnnounceListenManager/DegasServer public run {} {
     $self instvar send_alive_id_ 
     $self instvar alive_interval_
 
     $self announce_alive
     set send_alive_id_ [after $alive_interval_ "$self run"]
 }
 
 
 #------------------------------------------------------------------

 # AnnounceListenManager/DegasServer::log

 #

 # purpose : create a log message of the server.  This can be either

 #           to the screen or to a file.

 # input   : msg - a message to log

 # output  : none

 #------------------------------------------------------------------

 
 AnnounceListenManager/DegasServer private log {msg} {
     puts "$msg"
 }
 
 
 
 #------------------------------------------------------------------

 # AnnounceListenManager/DegasServer::get_program

 #

 # purpose : This is a util to convert the program submitted from a 

 #           client to a list, so that we can pass it to DegasAgent.

 # input   : addr,port - the client which submitted the program

 #           we are getting.

 # output  : none

 #------------------------------------------------------------------

 
 AnnounceListenManager/DegasServer private get_program {addr port} {
     $self instvar programs_
     return [array get programs_ $addr,$port,*]
 }