AvatarTransitionServer.sysml

// AVATAR CENTRALIZED TRANSITION SERVER %%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
package AvatarTransitionServer {
    private import Clocks::TimeOf;
    private import Time::TimeInstantValue;
    private import AvatarGeneral::*;
    private import AvatarCommunication::*;
    public import Occurrences::Occurrence;

	doc /* The server is used to select a transition to execute among the possible outgoing transitions 
	 * of a state of an Avatar block statemachine. For this, states will submit pools of requests to 
	 * the server, where each request is associated to one of its outgoing transitions. Then, as soon
	 * as some requests in the pool are feasible, the server chose one of them and send a response to
	 * the state identifying the selected request and providing the associated '@payload' (transmitted 
	 * message for receiving requests, and acknowledgment in other cases)
	 */
	 
// AVATAR TRANSITION REQUESTS ============================================

	abstract part def '#Request' {
		doc /* The most general request type. Outgoing transition from a state are statically ordered.
		 * Thus associated requests have an index w.r.t this order, which identifies the request they
		 * are associated to.
		 */
	
    	readonly attribute '@index' : Positive default 1;
    	
    	comment Request_Partition about '#OKrequest', '#NOKrequest', '#Request'
    	/* Requests are either certainly unfeasible (a #NOKrequest has false guard)
     	 * or potentially feasible (an #OKrequest has true guard)
     	 * #OKrequest, #NOKrequest is a partition of #Request
     	 */
    }
    
	part def '#NOKrequest' :> '#Request' {
		doc /* Certainly unfeasible requests, i.e. requests with false guard.
		 * they cannot be selected by the centralized server.
		 */
	}

	abstract part def '#OKrequest' :> '#Request' {
		doc /* Potentially feasible requests, i.e. requests with true guard.
		     * they may be selected by the centralized server.
   			 * They may be '@delay'ed by an "after", thus they have a trigger date w.r.t
   			 * the system discrete clock.
   			 */
	    
	    item localClock : Clock redefines localClock = '#systemClock' ;
	    readonly attribute '@delay' : Positive default 0;
	    derived attribute trigger_at = TimeOf(clock=localClock, o=self) + '@delay';

		comment OKrequest_Partition about '#OKrequest', '#TrivialRequest', '#CommunicationRequest'
    		/* OK requests may be without dependency and thus Trivialially feasible (#TrivialRequest). 
    		 * Non Trivialial requests are the ones associated to communication (#CommunicationRequest).
    		 * They may require synchronization, or channel or data to be available.
    		 * #TrivialRequest, #CommunicationRequest, is a partition of #OKrequest 
    		 */
	}
    
	part def '#TrivialRequest' :> '#OKrequest'{
		doc /* Trivialial requests don't communicate, thus don't block. Associated transitions
	     * only execute actions such as assignment, external method calls,...
	     */
	}
	
	item '#immediate_request' : '#TrivialRequest' = '#TrivialRequest'();
	
	abstract part def '#CommunicationRequest' :> '#OKrequest' {
		doc /* Communication requests, which have an associated communication channel.
		 * They may block until some ressource is available, depending of the kind of channel
		 */

    	readonly item '@channel' : '#Channel';
    	readonly attribute is_send : Boolean {
    		doc /* request direction, which is redoundant with #CommunicationRequest 
    		 * specializations but useful to factor some descriptions in the model
    		 */
		}

		comment about '#CommunicationRequest', '#SendRequest', '#ReceiveRequest'
		    /* communication requests have a direction: they are "send" or "receive". 
		     * #SendRequest, #ReceiveRequest, is a partition of #CommunicationRequest
		     */
    }

	part def '#SendRequest' :> '#CommunicationRequest' {
		doc /* Requests for sending. They have have a '@payload' (message) */
		
	    readonly item '@payload' : '#Message';
	    readonly attribute is_send redefines is_send = true;
	}
	
	part def '#ReceiveRequest' :> '#CommunicationRequest' {
		doc /* Requests for receiving */
		
	    readonly attribute is_send redefines is_send = false;
	}
	
	part def '#RequestPool' {
		doc /* Pools of requests. They contain one request per outgoing transition of the
		 * state they are associated to. They have an associated requestor identifying 
		 * who to send the response to (generally the block whose statemachine contains 
		 * the state)
		 */
		 
	    occurrence requestor : Occurrence;
	    item pool : '#Request'[0..*] ordered default := '#immediate_request';
	    
	    assert constraint {
	    	doc /* Request indexes correspond to their range in the pool. This is a redoundancy which
	    	 * makes the model more simple. 
	    	 * Moreover, all transitions from the concerned state in Avatar model have a corresponding 
	    	 * request , which cannot be expressed here as it is a constraint on the software which 
	    	 * generates the Avatar SysMLV2 model
	    	 */
	    	calc test {
	    	 	first start;
	    	 	then assign result := true;
	    	 	then for i in 1..size(pool) { 
	    	 		first start;
	    			then assign result := result & pool#(i).'@index' == i; 
	    		}
	    	}
	    	test()
	    }
		 
	}

// AVATAR RESPONSES TO REQUESTS ============================================
	item def '#Response' {
		doc /* Response to submitted request pool.
		 * Once a request in a pool is satisfied, the server sends a response to
	 	 * the requestor associated to the pool. This response identifies the
	 	 * (index of the transition corresponding to the) satisfied request in 
	 	 * the pool and carry some '@payload'. A '@payload' is a message (which can be
	 	 * a simple acknowledgement).
	 	 * Remark: As a resquestor should not have more than one pending request
	 	 * pool in the server, this information is sufficient to root the response
	 	 * to a unique well identified waiting accept action.
	 	 */
		
	    attribute '@index' : Positive default 1;
	    item '@payload' : '#Message' default '#ack_message';
	}
	item '#default_response' = '#Response'();

// THE SERVER ITSELF ========================================================

	part '#transitionRequestServer' {
		doc /* The transition request server. 
		 * General Principle: all submitted requests are initially considered as 
		 * '@delay'ed (even if their '@delay' is 0, for homogeneity). '@delay'ed requests
		 * wake up themself at their "trigger_at" date and launch the server's
		 * request handler on themself. The handler try to execute the request 
		 * which may lead either to successfull or blocking issue.
		 * A request wake up once and only once. After this, if it is blocked 
		 * (pending communication), it remains in the pending requests but is 
		 * removed from the '@delay'ed requests. Such requests may then be unblocked 
		 * by the wake-up of a dual request (a send for a receive or a receive for 
		 * a send), or discarded if some other request in their pool is selected 
		 * to be executed. Pools may also wait forever...
		 * 
		 * The server maintains two sets of pending pools:
		 * - general pending pools are all pending pools, containing all 
		 *   pending requests
		 * - '@delay'ed pending pools (each one associated to a general pending pools)
		 *   are pools containing '@delay'ed requests, and only '@delay'ed requests.
		 * 
		 * When a pool is submitted by a state, it is added to both sets. 
		 * When a '@delay'ed request wake up, it is removed from its '@delay'ed pool (which 
		 * is removed when becomming empty -- facultatif) and if handling 
		 * this request leads to a success, all involved pools are removed from
		 * both sets and relevant responses are send to the involved communicating 
		 * ends (which pools and which responses depends on channel types).
		 */
		 
		item localClock : Clock redefines localClock = '#systemClock' ;
		private individual item def Kill { doc /* type of the local kill signal */ }
		private individual item kill : Kill {
		 	doc /* The local kill signal to be sent to '@delay'ed requests when 
		 	 * their pool is deleted because some other request in the pool 
		 	 * is executed
		 	 */
		}
		private item mutex : '#Lock';
    	private item request_pools : '#RequestPool'[0..*] ordered {
    	 	doc /* general pending pools containing requests waiting for wake up 
    	 	 * or available ressources
    	 	 */
    	}
   		protected item delayed_pools : '#delayedPool'[0..*] ordered {
    		doc /* '@delay'ed pending pools containing requests waiting for wake up */
    	}
		
		part def '#delayedPool' {
			doc /* Type of pools containing '@delay'ed requests. Such pools are
			 *  associated to the general request pool that contain their requests.
			 */
    		item request_pool : '#RequestPool';
    		item delayed_requests : '#delayedRequest'[0..*] ordered;
		}

		part def '#delayedRequest' {
			doc /* (operational) structure for '@delay'ed requests */

    		attribute trigger_at : Positive, TimeInstantValue { doc /* The date at which to wake up */ }
    		item delayed_pool : '#delayedPool' { doc /* the '@delay'ed pool the request belongs to */}
    		attribute '@index' : Positive;
    		
    		private readonly item myself = self; // technical... to be referenced in sub-components
		    protected item delayed_pools : '#delayedPool'[0..*] ordered = '#transitionRequestServer'.delayed_pools;
		    // apparently required !?!
    
		    state statemachine { 
		    	doc /* waits for wake up or kill */
		        entry; then state wait;
		        transition first wait accept Kill then done;
		        transition first wait accept at trigger_at
		          do action {
		            perform action remove_request subsets 
		            	'#multiset_remove' {
		                	in multiset redefines multiset = delayed_pool.delayed_requests; 
		                    in pred redefines pred = '#identity'(myself); 
		                 }
                    then if delayed_pool.delayed_requests == null {
 		            	 	perform action remove_delayed_pool subsets 
		            		'#multiset_remove' {
		                		in multiset redefines multiset = delayed_pools; 
		                    	in pred redefines pred = '#identity'(delayed_pool); 
		                 	} then done;
		                 }
		            then action wake_up subsets '#transitionRequestServer'.request_wake_up { 
		                        in request_pool redefines request_pool = delayed_pool.request_pool; 
		                        in transition_id redefines transition_id = transition_id; 
		                 }
           		  }
				  then done;
		    }
		    first statemachine then done;
		}
		
		// FOR SUBMITTING A REQUEST POOL ==========================================

		action submit { // add parameter pool to request_pools and delayed_pools
			in request_pool : '#RequestPool';
			part delayed_pool : '#delayedPool' { 
     			item request_pool : '#RequestPool' redefines request_pool = request_pool;
    			item delayed_requests : '#delayedRequest'[0..*] redefines delayed_requests := null;
    		}
    		first start; then perform mutex.lock;
    		for i in 1..size(request_pool.pool) {
    			if request_pool.pool#(i) hastype '#OKrequest' { // build '@delay'ed request for ith request
    				item request := request_pool.pool#(i) as '#OKrequest';
    				part delayed_request : '#delayedRequest' {
    					in trigger_at redefines trigger_at = request.trigger_at;
    					in delayed_pool redefines delayed_pool = delayed_pool;
    					in '@index' redefines '@index' = request.'@index';
    				}
    				first start; then // add '@delay'ed request
    				assign delayed_pool.delayed_requests := (delayed_pool.delayed_requests,delayed_request);
    				then delayed_request.start;
    			}
    		} 
    		then assign request_pools := (request_pools, request_pool); // add general request pool
    		then if delayed_pools.delayed_requests != null {
    			assign delayed_pools := (delayed_pools, delayed_pool);
    		} // add '@delay'ed request pool if not empty
    		then perform mutex.unlock;
    		then done;
		}
		
		// REQUEST WAKE UP: tries request synchronization if request has not already been handled
		
		comment about request, request_pool /* ALGORITHMIC VARIABLES: inputs for trying a synchronization */
    	protected part request : '#Request'{ doc /* ALGORITHMIC VARIABLE : the request to be synchronized */ }
    	protected part request_pool : '#RequestPool'{ doc /* tALGORITHMIC VARIABLE :he pool of the request to be synchronized */ }

		action request_wake_up { 
			doc /* verifies that request has not already been synchronized and if not, try to synchronize it */
		    in request_pool : '#RequestPool'; 
		    in transition_id : Positive;
		    
		    first start; then perform mutex.lock;
		    then perform action todo subsets '#multiset_exists'{ 
		    	in multiset redefines multiset = request_pools; 
		    	in pred redefines pred = '#identity'(request_pool); }
		    then if todo.result { 
		    	first start;
		    	then assign request := request_pool#(transition_id);
		    	then assign request_pool := request_pool;
		    	then perform try_request_synchronization;
		    }
		    then perform mutex.unlock;
		    then done;
		}

        action try_request_synchronization {
            first start;
            then perform search_request_synchronization;
            then perform remove_request_pools;
            then perform send_mails;
            then done;
		}
            
		// SUBROUTINE : removing synchronized pending pools =========================
		
		item pools_to_remove : '#RequestPool'[0..2] { 
			doc /* ALGORITHMIC VARIABLE: synchronized pools, i.e associated to synchronized
			 * requests. Updated when finding a synchronization. Removed from pending  pools
			 * when executing this synchronization.
			 */
		}

	    protected action remove_request_pools { 
	    	doc /* Removing pools in pools_to_remove, i.e removing them from both sets of
	    	 * pending pools (general and '@delay'ed). This also kills the process in '@delay'ed 
	    	 * requests which are waiting for either a trigger signal or a kill signal.
	    	 */
	    	 
	        // subroutine
            action kill_delayed_pool_requests : '#Unary_action' { 
            	doc /* kill all waiting requests of a parameter '@delay'ed pool */
                in parameter : '#delayedPool';
                first start;
                then for r in parameter.delayed_requests { first start; then send kill to r; }
            }

			first start;
	        then for p : '#RequestPool' in pools_to_remove {
	        	
	        	// subroutine
		        calc has_request_pool : '#Unary_predicate' { 
		        	doc /* test if the request pool of a parameter '@delay'ed pool is p */
		        	in parameter : '#delayedPool'; parameter.request_pool == p
		        }
		        
	            first start;
	            then action remove_request_pool subsets '#multiset_remove' { // remove pool from general pending pools
	                in multiset redefines multiset = request_pools; 
	                in pred redefines pred = '#identity'(p); 
	            }
				then action remove_delayed_pool subsets  '#multiset_iter' { 
	                in multiset redefines multiset = delayed_pools; 
	                in pred redefines pred = has_request_pool; 
	                in if_act redefines if_act = kill_delayed_pool_requests; 
	            }
	            succession flow remove_delayed_pool.discarded to update_delayed_pools.new_delayed_pools;
	            action update_delayed_pools {
	                in new_delayed_pools  : '#RequestPool'[0..*];
	                first start; then assign delayed_pools := new_delayed_pools; then done;
	            }
	        }
	        then done;
	    }
		
		// SUBROUTINE: sending responses to requestors =============================

	    protected part def '#Mail' { 
	    	doc /* Response computed while finding synchronizations have an associated recipient,
	    	 * This characterizes mails.
	    	 */
	        item recipient: Occurrence;
	        item response: '#Response';
	    }
	    
        item mails_to_send : '#Mail'[0..2] {
        	doc /* ALGORITHMIC VARIABLE: mails associated to synchronized requests. Updated 
        	 * when finding a synchronization. Responses are sent to recipients when executing 
        	 * this synchronization.
			 */
        }

    	protected action send_mails {
    		doc /* send responses to requestors of synchronized requests */
    		first start;
        	then for i : Positive in 1..size(mails_to_send) {
            	first start;
            	then send mails_to_send#(i).response to mails_to_send#(i).recipient;
            	then done;
        	}
        }

	    // SUBROUTINE searching a dual request for current request in pending requests ===============
	
	    private part dual_request : '#Request'[0..1]{
	    	doc /* selected dual pending request of the current request, 
	         * among the found ones
	         */
	    }
	    private part dual_request_pool : '#RequestPool'[0..1] {
	    	doc /* ALGORITHMIC VARIABLE: the pool containing the selected dual pending 
	    	 * request of the current request.
	    	 */
	    }
	
	    action search_pending_dual_request {
		    doc /* search for a dual request of current request in request_pools.
		     * updates dual_request and dual_request_pool accordingly. If no dual request 
		     * is found, they are set to null. If severall are found, one is choosed
		     * randomly.
		     * Dual requests are communication requests on a same channel with opposite 
		     * directions
		     */
			
	        comment about dual_requests, dual_request_pools
	            /* order in these two items are parallel: i-th request belongs to i-th pool */
	
	        part dual_requests : '#Request'[0..*] := null{
	        	doc /* found requests that are dual of current request */
			}
	        
	        part dual_request_pools : '#RequestPool'[0..*] := null {
	        	doc /* the pools containing dual requests in dual_requests, in the same order */
	        }
	               
	        first start;
	        then assign dual_requests := null;
	        then assign dual_request_pools := null;
	        then decide;
	        	if request hastype '#CommunicationRequest' then search; 
	        	else done;
	        	
	        item communication_request[0..1] : '#CommunicationRequest' := request as '#CommunicationRequest';
	        
	        action search assign dual_request := null;
	        then assign dual_request_pool := null;
	        then for p : '#RequestPool' in request_pools {
	            item found : '#Request'[0..*] := null; // dual requests in pool p
	            first start;
	            then for r : '#Request' in p.pool {
	            	if r hastype '#CommunicationRequest' {
	            		item req = r as '#CommunicationRequest';
	            		first start;
	                	then if communication_request.'@channel' == req.'@channel' and 
	                        	communication_request.is_send != req.is_send and
	                        	communication_request.trigger_at >= localClock.currentTime
	                     { first start; then assign found := (found,r); then done; }
	                }
	            }
	            then if found != null {
	            	first start;
	                then assign dual_request_pools := (dual_request_pools, p);
	                then assign dual_requests := (dual_requests,'#multiset_random'(found));
	                then done;
	            }
	            then done;
	        }
	        then if dual_requests != null { // randomly select one of the possible dual requests
	        		attribute x : Positive := '#bound_random'(low=1, high=size(dual_requests));
	            	first start;
	            	then assign dual_request_pool:= dual_request_pools#(x);
	                then assign dual_request:= dual_requests#(x);
	             }
	             else {
	                first start;
	                then assign dual_request_pool:= null;
	                then assign dual_request:= null;
	             }
	        then done;
	    }
	
	    // SUBROUTINE add an acknowledgement for a couple (pool,request)
	    
	    action acknowledge {
	    	doc /* add acknowledgement in mails_to_send and pools_to_remove */
	        in request_pool : '#RequestPool';
	        in request : '#Request';
	        first start;
	        then assign mails_to_send := ( mails_to_send,
	        	'#Mail'( recipient = request_pool.requestor,
	                     response = '#Response'('@index' = request.'@index',
	                                            '@payload' = '#ack_message') ) );
	        then assign pools_to_remove := (pools_to_remove, request_pool);
	    }
	
	    // SEARCH A SYNCHRONIZATION FOR request ==============================
	    
	    action search_request_synchronization { 
	    	doc /* search a synchronization for current request and updates mails_to_send
	    	 * and pools_to_remove accordingly. If no synchronization is found, both
	    	 * variables are set to null.
	    	 */
			first start;
			then assign mails_to_send := null;
			then assign pools_to_remove := null;
	        then decide select_wrt_request_type;
	            if request hastype '#NOKrequest' then done;
	            if request hastype '#TrivialRequest' then trivial_synchro;
	            else communication_synchro;
			
	        // Synchronization of Trivialial actions --------------------------------
	        action trivial_synchro subsets acknowledge {
	        	doc /* Trivialial actions are always synchronized (with no other pool) and simply return 
	        	 * an acknowledgment
	        	 */
	        	in request_pool redefines request_pool;
	        	in request redefines request;
	        }
	        then done; // end of trivial_synchro : search_request_synchronization returns
	        
	        // Synchronization of communication requests, depends on channel type ----
	        item communication_request[0..1] : '#CommunicationRequest' := request as '#CommunicationRequest';
	        decide communication_synchro;
	            if communication_request.'@channel' hastype '#Fifo' then fifo_synchro;
	            else sync_synchro;
	            
	        // Synchronization of communication requests on fifo channel ------------
	        action fifo_synchro {
	        	item fifo : '#Fifo' := communication_request.'@channel' as '#Fifo';
	        	first start;
	            then decide select_wrt_communication_direction;
	                if request hastype '#SendRequest' then send_synchro;
	                else receive_synchro;
	            
	            action send_synchro {
	            	item send_request : '#SendRequest' := communication_request as '#SendRequest';
	            	first start;
	                then decide test_if_lossy;
	                    if fifo.'@relation'.'@lossy' == true then lossy_synchro;
	                    else noloss_synchro;
	                
	                decide lossy_synchro; // loss or not, random choice
	                    then loss_synchro;
	                    then noloss_synchro;
	                    
	                action loss_synchro subsets acknowledge {
	                	in request_pool redefines request_pool = request_pool;
	                	in request redefines request = request;
	                }
	                then done;
	                
	                action noloss_synchro {
	                	first start;
	                    then decide test_if_full;
	                        if fifo.can_put() then put_synchro;
	                        else full_synchro;
	                    
	                    decide full_synchro; // test if blocking
	                        if fifo hastype '#NBfifo' then discard_synchro;
	                        else done;
	                               
		                action discard_synchro subsets acknowledge {
		                	in request_pool redefines request_pool = request_pool;
		                	in request redefines request = request;
		                }
		                then done;
		                
	                    action put_synchro {
	                        first start;
	                    	then action put subsets fifo.put {
	                    		in 'message' redefines 'message' = send_request.'@payload';
	                    	}
	                        then action ack subsets acknowledge {
	                			in request_pool redefines request_pool = request_pool;
	                			in request redefines request = request;
	                		}
	                        then perform search_pending_dual_request;
	                        then
	                        if dual_request != null {
	                            first start;
								then assign pools_to_remove := (pools_to_remove,dual_request_pool);
	                            then assign mails_to_send := (mails_to_send,
	                                 '#Mail'( recipient = dual_request_pool.requestor,
	                                          response = '#Response'( '@index' = dual_request.'@index',
	                                                                  '@payload' = fifo.get() )));
	                            then done;
							}
	                    } then done;
	                } then done; 
	            } then done; // end of send_synchro : fifo_synchro returns
	            
	            action receive_synchro {
	                first start;
	                then if fifo.can_get() {
	                    first start;
	                    then assign mails_to_send := 
	                         '#Mail'( recipient = request_pool.requestor,
	                                 response = '#Response'('@index' = communication_request.'@index',
	                                                        '@payload' = fifo.get() ));
	                    then assign pools_to_remove := request_pool;
	                    then perform search_pending_dual_request;
	                    then
	                    if dual_request != null {
	                        first start;
	                        then assign pools_to_remove := (pools_to_remove,dual_request_pool);
	                        then action ack subsets acknowledge {
	                            in request_pool redefines request_pool = dual_request_pool;
	                            in request redefines request = dual_request;
	                        } then done;
	                    } then done;
	                } then done;
	            } then done; // end of receive_synchro : fifo_synchro returns
	        } then done; // end of fifo_synchro : search_request_synchronization returns
	
	        // Synchronization of communication requests on synchroneous channel ------------
	        action sync_synchro {
	            first start;
	            then decide select_wrt_communication_direction;
	                if communication_request hastype '#SendRequest' then send_synchro;
	                else receive_synchro;
	            
	            action send_synchro {
	                first start;
	                then perform search_pending_dual_request;
	                then
	                if dual_request != null {
                    	item send_request : '#SendRequest' = request as '#SendRequest';
	                    first start;
	                    then action ack subsets acknowledge {
	                        in request_pool redefines request_pool = request_pool;
	                        in request redefines request = request;
	                    }
                        then assign mails_to_send := ( mails_to_send,
                                    '#Mail'( recipient = dual_request_pool.requestor,
                                            response = '#Response'(
                                                '@index' = dual_request.'@index',
                                                '@payload' = send_request.'@payload' )) );
                        then assign pools_to_remove := (pools_to_remove, dual_request_pool);
                        then done;
                    } then done;
	            } then done; // end of send_synchro : sync_synchro returns
	            
	            action receive_synchro {
	                first start;
	                then perform search_pending_dual_request;
                    then
                    if dual_request != null {
                    	item send_request : '#SendRequest' = dual_request as '#SendRequest';
                        first start;
                        then action ack subsets acknowledge {
                            in request_pool redefines request_pool = dual_request_pool;
                            in request redefines request = dual_request;
                        }
                        then assign mails_to_send := ( mails_to_send,
                                    '#Mail'( recipient = request_pool.requestor,
                                            response = '#Response'(
                                                '@index' = request.'@index',
                                                '@payload' = send_request.'@payload' )) );
                        then assign pools_to_remove := (pools_to_remove, request_pool);
                        then done;
                    } then done;
	            } then done; // end of receive_synchro : sync_synchro returns
	        } then done; // end of sync_synchro : search_request_synchronization returns
	    } // end of search_request_synchronization
	}
}