<?php/***************************************************************************************\* lib/base_lib.php ** Contains the IPC class and functions that need to be avalible across ** the project. No views should be defined inside of this, create a ** seperate lib file. The seperation between the view an controler should ** be sufficient to do anything you need to do. *\***************************************************************************************/class dtenix_ipc{ //member varibles. private $shm_key; //shared memory key private $shm_id; //shared memory id private $sem_id; //semaphore id private $msg_queue; //our msg queue //keys (cannot be flagged) const SEM_KEY = 1; //semaphore key const MSG_KEY = 2; //message queue key const DEF_KEY = 1337; //key for the defined varibles //message types (cannot be flagged) const MSG_TO_CONT = 1; //message to controler const MSG_DOWN = 2; //message to all child processes and views that deamon is halting const MSG_DOWN_ACK = 3; //message from child proccess that they have detached from SHM const MSG_TO_NEW_VIEW = 4; //message from controler to a view that is requesting register const MSG_TO_VIEW_BASE = 200; //base value for registered views. Add the view identity to this. const MSG_TO_PORT_BASE = 500; //base value for the msg to ports. Add the port identity to this. //errors const E_VARNOSET = 2; //No shared varible of that name defined const E_VARNOMEM = 4; //well crap... out of system memory //misc const MSG_MAX_SIZE = 256; //our largest message size... todo fixor this ugly hack /* * __construct($msg_type) * yay for constructors. Sets up the memory stuff. Make sure you try { }catch around it. * $msg_type used to set up what kind of object this is. */ public function __construct($msg_type) { //lets get our shared memory if ( ($this->shm_key = ftok(dirname(__FILE__),'a')) === false ) //get our shm_key { throw ( $error = new Exception("Failed to get Shared memory key.") ); } if ( ($this->sem_id = sem_get(self::SEM_KEY)) === false ) //get our semaphore ID { throw ($error = new Exception("Failed to get semaphore key.") ); } if ( ($this->shm_id = shm_attach($sem_id, self::TIMEOUT)) === false ) //get the id for the SHM block { throw ( $error = Exception("Failed to attach to the SHM block.") ); } if ( ($this->msg_queue = msg_get_queue( self::MSG_KEY,0600)) === false) //get our message queue { throw ($error = Exception("Failed to get message queue.") ); } $this->msg_to_self = $msg_type; //set up our message type } /* * get_var((string)$varname) * Returns the value of an already declared shared varible declared as $varname in an array * $retval[$varname] == 'value of the shared varible' * If no shared variable of that name has been declared, or there is an error getting the semaphore * it tosses an error. */ //todo typecast $varname here and in set_var public function get_var($varname) { $retval = null; //what we are going to return $thekey = null; //key to the varible, returned from $this->check_var_exists() //get the semaphore if ( sem_get($this->sem_id) === false) { throw ($error = new Exception("Could not acquire semaphore") ); } if ( ($thekey = $this->check_var_exists($varname)) !== false ) { //defined, set in $retval[$varname] $retval = array($varname=> shm_get_var($this->shm_id,$thekey)); } //release the semaphore if ( sem_release($this->sem_id) === false ) { throw ($error = "Could not release semaphore."); } //check to make sure $retval has been set if ( $retval == null ) { //not defined, toss error throw($error = self::E_VARNOSET); } return $retval; } /* * set_var((string)$varname, $set_value) * * If no shared variable of that name has been declared, there is an error getting the semaphore, * or the system runs out of shared memory, it tosses an error. */ public function set_var($varname, $set_value) { $retval = self::E_VARNOSET; //checked $thekey = null; //key to the varible, returned from $this->check_var_exists() if ( sem_get($this->sem_id) === false) { throw ($error = "Could not acquire semaphore"); } if ( ($thekey = $this->check_var_exists($varname)) !== false ) { //defined, set in $retval[$varname] $retval = array($varname=> shm_put_var($this->shm_id,$thekey, $set_value)); } //release the semaphore if ( sem_release($sem_id) === false ) { throw ( $error = new Exception("Could not release semaphore") ); } //check to make sure $retval has been set switch ( $retval ) { case self::E_VARNOSET: //no shared varible by that name throw ( $error = new Exception("Retreval error: ". self::E_VARNOSET) ); break; case false: //out of system memory throw ( $error = new Exception("Retreval error: ". self::E_VARNOMEM) ); break; case true : //everything worked fine return true; break; /* case self::E_PORNOVERFLOW: //sticky page file. throw ( $error = self::E_PORNOVERFLOW ); break; */ } return false; //wtf..?! } /* * check_var_exists( (string)$varname) * Only to be called when the semaphore $this->sem_id is aquired * Checks to see if a varible called $varname has been registered * by the controler and returns the key if it has. Returns false if unregistered. */ //todo typecast $varname as string private function check_var_exists($varname) { $d_arr = shm_get_var($this->shm_id, self::DEF_KEY); if ( isset($d_arr[$varname]) ) { return $d_arr[$varname]; } return false; } /* * tell_controler($msg, &$errno, $block = true) * sends the message $msg to the controler process. If the process is such that it * cannot afford to block, pass false as the optional second parameter. Note that * the call will only block if there is no more space in the message queue. * This function will return false if there was an error sending the message, and * set $errno to the value of the error. It would be a good idea to check that. */ public function tell_controler($msg, &$errno, $block= true) { $errno = null; //if $block === false and the msg_send fails, it's good to know why if( ! msg_send($this->msg_queue, self::MSG_TO_CONT, $msg, true, $block, $errno) ) { //This point will only be reached if $block is false and the queue is full, //or if msg_send sends a doubleplusungood message. if ( $errno === EAGAIN ) //see if the queue wants us to try again... { if ( !msg_send($this->msg_queue, self::MSG_TO_CONT, $msg, true, $block, $errno) ) { return false; //darnit, didn't work twice. exiting. } } else { return false; //ok something bad happened. } } return true; } /* * listen_for_msg( &$msg, $block = true) * Check the message queue for a message to this process. If a message exists $msg will be * filled out and the function will return true. If the optional parameter $block is set to * false, the function will simply check to see if a message exists and will return false if * not. * * **WARNING** * By default this function blocks untill a message comes in. If a child process does this * in a manner that doesn't assure that the message queue is set up first, then it will * block indefinately causing the process to hang. Be careful. * **END WARNING MMKAY?** * */ public function listen_for_msg( &$msg, $block = true) { $errno = null; //if the funtion fails, this is set to why $msg_type = $this->msg_to_self; //message type received if ($block == false) { $block = MSG_IPC_NOWAIT; //set up the flags } else { $block = 0; // no blocking } if (! msg_receive($this->msg_queue, $this->msg_to_self, $msg_type, self::MSG_MAX_SIZE, $msg, true, $block, $errno) ) { $msg = $errno; return false; //problem with reception or no message waiting (if $block was false) } return true; } /* * function change_msg_type( int $new_type) * Changes the type of message that $this->listen_for_msg() listens for. In general * this is a bad idea, but is neccisary for self attaching proccesses. */ protected function change_msg_type(int $new_type) { $this->msg_to_self = $new_type; } /* * __destruct() * class destructor... sends the message to the queue that this object is shutting down * Note, this does not clean up the shared memory. The controler class needs to do that. */ public function __destruct() { msg_send($this->msg_queue,MSG_DOWN_ACK,'shutting it down'); sem_release($this->sem_id); //More than likely this will cause an warning but just in case... shm_detach($this->shm_id); //Detach from shared memory. Remember that the memory is still assigned. //It is the controlers resposibility to clear that up. }}/* * class dtenix_ipc_port * implementation of dtenix_ipc for the port management processes * not much needed right this second, this is more for future proofing */class dtenix_ipc_port extends dtenix_ipc{ public function __construct($port_num) { $port_num = dtenix_ipc::MSG_TO_PORT_BASE + $port_num; //set up the port message type. parent::__construct($port_num); //init everything. } public function __destruct() { parent::__destruct(); }}/* * class dtenix_ipc_view * implementation of dtenix_ipc for the view process */class dtenix_ipc_view extends dtenix_ipc{ //consts //view states const VIEW_UNREG = 1; const VIEW_REG = 2; private view_state; public function __construct() { parent::__construct(dtenix_ipc::MSG_TO_NEW_VIEW); //init base functions $msg_error = null; //error message if the send/recev of a msg fails $this->view_state = self::VIEW_UNREG; //init our view state correctly. $this->register_view(); //register our view with the controler } /* * register_view() * Sends a message to the controler letting it know that we are here and looking for * view updates. Throws an exception if the register failed, false if we are registered already * and true if registration was successfull. */ public function register_view() { $msg = null; //the message we receive from the controler if ( $this->view_state == self::VIEW_REG ) //are we registered already? { return false; } //tell the controler that we are here. if ( ! $this->tell_controler(dtenix_ipc::MSG_TO_NEW_VIEW, $msg_error)) { //suckage. message queue either backed up or something is really fucked up. throw ($error = new exception("Send message failure ". $msg_error)); } //try this for 3 times, not blocking in case the controler is locked up or not running for ($looper = 0; $looper <4; $looper++) { if ( $this->listen_for_msg($msg,false) ) { $this->view_state = self::VIEW_REG; $this->change_msg_type($msg); return true; } elseif ( $msg !== 42 ) //if the problem wasn't no message waiting we have an issue. { throw ($error = new exception('Receive error '. $msg); } sleep(1); } throw ($error = new exception($msg); //failed to register with the controler } /* * unregister_view() * tells the controler that we are closing out. * throws an exception if there was an error, which would only happen in really bad situations. */ public function unregister() { $error = null if ( !$this->tell_controler(dtenix_IPC::MSG_UNREG,$error) ) { //major bad mojo... this should never fail. throw ($error = new exception('unreg error'.$error); } } public function __destruct() { $this->unregister_view(); parent::__destruct(); }}/* * class dtenix_ipc_control * implementation of the dtenix_ipc class for the controler process */class dtenix_ipc_control extends dtenix_ipc{ public function __construct() { parent::__construct(dtenix_ipc::MSG_TO_CONT); } /* * make_var( string $var_name) * creates a varible with the name $var_name and the value of $value * value defaults to null so that we don't get garbage values if * child proccesses try to read from it before it's actually used. */ public function make_var( string $var_name, $value = null ) { $d_arr = null; //holder varible so we can re-form the key list $error = true; //gotta figure out a way to //get the semaphore if ( sem_get($this->sem_id) === false) { throw ($error = "Could not acquire semaphore"); } //check to see if the varible exists in the key registry already if ( $this->check_var_exists($var_name) !== false ) { $error = new Exception("Varible exists already."); } else { $d_arr = shm_get_var($this->shm_id, dtenix_ipc::DEF_KEY); $d_arr[$var_name] = $value; if ( !shm_put_var($this->shm_id, dtenix_ipc::DEF_KEY, $d_arr) ) { $error = new Exception("Unable to reset key"); } } if ( sem_release($this->sem_id) === false) { throw ( $error = new Exception("Unable to release semaphore.") ); } if ( is_a($error, Exception) ) { throw $error; } return true; } /* * __destruct() * Called automatically when the object is closed out. It needs send out * self::MSG_DOWN messages one at a time untill all registered child processes * have responded with a self::MSG_DOWN_ACK message. After that it clears all * self::MSG_TO_CONT messages out of the queue detaches from shared memory * and from the queue, and destroying both of them. */ public function __destruct() { //todo this. waiting on list of registered processes }}?>