File Source for threads_manager.class.php

<?php
// *** Tangra (Application Framework and Tools for PHP)
// $Id$
//
/**
* Contains class Threads_Manager
*
* @package tangra_lib
* @subpackage web_site
*/
/**
*
*/
require_once(TANGRA_MAIN_DIR.'core/vars_manager.class.php');
/**
*
*/
require_once(TANGRA_MAIN_DIR.'web_site/session_vars_manager.class.php');
/**
*
*/
require_once(TANGRA_MAIN_DIR.'web_site/thread_vars_manager.class.php');
/**
* Variable name for TVM
*
*/
define('THREADS_MANAGER_NAME', '_tm');
//TODO - notify on expire to add methods
/**
* Threads Manager
*
* Threads concept is enhancement over sessions. Sites that implements threads management
* can detect when new instance of browser is opened and user is working with two or more
* browsers simultaneously.
* Threads manager function is to detect such "forks" and to split dataspace for the treads
* so each tread to have separate dataspace.
* This class is not ment to be used directly by the users - quite the opposite - it is ment to be
* 100% transparent. User have to use Web_Site/Web_Page get_tvm methods in order to get TVM.
*
* @package tangra_lib
* @subpackage web_site
* @see Thread_Vars_Manager
*
*/
class Threads_Manager extends Tangra_Class {
/**
* Default lifetime for snapshots
*
*/
const SNAPSHOTS_DEFAULT_LIFETIME = 300;
/**
* Default maximum count of snapshots. Intended to prevend DoS attacks
*
*/
const MAX_SNAPSHOTS = 250;
/**
* name that will be used for the variable that will hold thread "id"
*
* @var string
* @internal
*/
private $url_rewrite_var_name;
/**
* Array that contain all existing threads
*
* @var mixed
* @internal
*/
private $threads = array();
/**
* Array that contain map url vars to thread index
*
* @var array
* @internal
*/
private $url_var_to_thread_links = array();
/**
* Index of current thread
*
* @var integer
* @internal
*/
private $current_thread_index;
/**
* Current URL variable (thread "id")
*
* @var string
* @internal
*/
private $current_url_var;
/**
* Snapshots of threads
*
* @var array
* @internal
*/
private $threads_snapshots = array();
/**
* Lifetime for snapshots in seconds
*
* @var integer
*/
private $snapshots_lifetime;
/**
* Maximum count of snapshots.
*
* @var integer
*/
private $max_snapshots;
/**
* Holds next unused thread index
*
* @var integer
*/
private $next_thread_index = 0;
/**
* Holds next unused snapshot index per thread
*
* @var array
*/
private $next_snapshot_index = array();
/**
* Constructor
*
* @param unknown_type $url_rewrite_var_name Name for the URL variable that will be used to track threads
* @param unknown_type $snapshots_lifetime Number of seconds after which snapshot will be seen as garbage and eventually unset
*/
function __construct($url_rewrite_var_name, $snapshots_lifetime = Threads_Manager::SNAPSHOTS_DEFAULT_LIFETIME, $max_snapshots = Threads_Manager::MAX_SNAPSHOTS) {
$this->url_rewrite_var_name = $url_rewrite_var_name;
$this->snapshots_lifetime = $snapshots_lifetime;
$this->max_snapshots = $max_snapshots;
}
/**
* Processes current context and manages threads
*
* @param Web_Context $context
*/
public function process(Web_Context $context) {
$this->new_pass();
if ($this->is_url_rewrite_var_passed($context)) {
$incoming_url_var = $this->get_url_rewrite_var($context);
if (array_key_exists($incoming_url_var, $this->url_var_to_thread_links)) {
if ($this->url_var_to_thread_links[$incoming_url_var]['access_count'] == 0) {
$this->url_var_to_thread_links[$incoming_url_var]['access_count']++;
//continue
$this->current_thread_index = $this->url_var_to_thread_links[$incoming_url_var]['thread'];
//printbr('continue '.$this->current_thread_index);
} else {
//fork
$this->create_new_thread();
//printbr('fork '.$this->get_current_thread_index());
$this->move_snapshot($this->url_var_to_thread_links[$incoming_url_var]['thread'],
$this->url_var_to_thread_links[$incoming_url_var]['snapshot'],
$this->get_current_thread_index(),
$incoming_url_var
);
}
} else {
$this->create_new_thread();
//printbr('wrong');
}
} else {
$this->create_new_thread();
//printbr('new_direct');
}
// $this->collect_garbage();
}
/**
* Updates last access time of current (detected from context) snapshot
*
* @param Web_Context $context
*/
public function snapshot_update_access_time(Web_Context $context) {
if ($this->is_url_rewrite_var_passed($context)) {
$incoming_url_var = $this->get_url_rewrite_var($context);
if (array_key_exists($incoming_url_var, $this->url_var_to_thread_links)) {
$current_thread_index = $this->url_var_to_thread_links[$incoming_url_var]['thread'];
$current_snapshot_index = $this->url_var_to_thread_links[$incoming_url_var]['snapshot'];
$this->threads_snapshots[$current_thread_index][$current_snapshot_index]['last_access_time'] = time();
//updates (if exist) access time of previos snapshot. That way we can correctly handle browser refresh button
if (array_key_exists($current_snapshot_index - 1, $this->threads_snapshots[$current_thread_index])) {
$this->threads_snapshots[$current_thread_index][$current_snapshot_index - 1]['last_access_time'] = time();
}
//printbr("Updating: $current_thread_index $current_snapshot_index $incoming_url_var");
}
}
}
/**
* Returns lifetime setting for snapshots
*
* @return integer Lifetime in seconds
*/
public function get_snapshots_lifetime() {
return $this->snapshots_lifetime;
}
/**
* Creates new thread
* @internal
*
*/
private function create_new_thread() {
$this->current_thread_index = $this->add_new_thread();
}
/**
* Saves snapshot ot current thread
*
*/
public function save_snapshot() {
$current_url_var = $this->get_current_url_var();
$current_thread_index = $this->get_current_thread_index();
$this->url_var_to_thread_links[$current_url_var]['thread'] = $current_thread_index;
$this->url_var_to_thread_links[$current_url_var]['access_count'] = 0;
$snapshot_index = $this->add_snapshot($current_thread_index);
$this->url_var_to_thread_links[$current_url_var]['snapshot'] = $snapshot_index;
}
/**
* Moves snapshot from one thread to another.
*
* @param integer $old_threat_index
* @param integer $snapshot_index
* @param integer $new_thread_index
* @param string $url_var
* @internal
*/
private function move_snapshot($old_threat_index, $snapshot_index, $new_thread_index, $url_var) {
$snapshot = $this->get_snapshot($old_threat_index, $snapshot_index);
$this->threads[$new_thread_index] = clone $snapshot['snapshot'];
$this->url_var_to_thread_links[$url_var]['thread'] = $new_thread_index;
$this->url_var_to_thread_links[$url_var]['access_count'] = 1;
//printbr('Moving '.$old_threat_index.' '.$snapshot_index.' to '.$new_thread_index.' '.$url_var);
$snapshot_index_new = $this->add_snapshot($new_thread_index);
$this->url_var_to_thread_links[$url_var]['snapshot'] = $snapshot_index_new;
}
/**
* Prepares ovject for new process
* @internal
*/
private function new_pass() {
$this->current_url_var = $this->get_new_url_var();
output_add_rewrite_var(THREADS_MANAGER_NAME, $this->get_current_url_var());
}
/**
* Adds snapshot for thread specified by <var>$thread_num</var>
*
* @param integer $thread_num thread index
* @return integer Snapshot index
* @internal
*/
private function add_snapshot($thread_num) {
if ($this->count_snapshots() > Threads_Manager::MAX_SNAPSHOTS) {
throw new Tangra_Exception('Maximum count of snapshots ('.Threads_Manager::MAX_SNAPSHOTS.') reached. Your session is considered DoS attack.');
}
if (!array_key_exists($thread_num, $this->next_snapshot_index)) {
$c = 0;
} else {
$c = $this->next_snapshot_index[$thread_num];
}
$this->next_snapshot_index[$thread_num] = ++$c;
$this->threads_snapshots[$thread_num][$c]['create_time'] = time();
$this->threads_snapshots[$thread_num][$c]['snapshot'] = clone $this->threads[$thread_num];
$this->threads_snapshots[$thread_num][$c]['last_access_time'] = $this->threads_snapshots[$thread_num][$c]['create_time'];
$this->collect_garbage();
return $c;
}
private function count_snapshots() {
$ret = 0;
foreach ($this->threads_snapshots as $thread) {
$ret += count($thread);
}
return $ret;
}
/**
* Returns snapshot of thread
*
* @param integer $thread_num thread index
* @param unknown_type $snapshot_num Snapshot index
* @return array
* @internal
*/
private function get_snapshot($thread_num, $snapshot_num) {
$ret = false;
//printbr('Getting snapshot '.$thread_num.' '.$snapshot_num);
if (array_key_exists($snapshot_num, $this->threads_snapshots[$thread_num])) {
$ret = $this->threads_snapshots[$thread_num][$snapshot_num];
//printbr('OK');
}
return $ret;
}
/**
* Returns current thread index
*
* @return integer
*/
public function get_current_thread_index() {
return $this->current_thread_index;
}
/**
* Returns current URL variable value
*
* @return unknown
*/
public function get_current_url_var() {
return $this->current_url_var;
}
/**
* Returns URL variable name
*
* @return unknown
*/
public function get_url_rewrite_var_name() {
return $this->url_rewrite_var_name;
}
/**
* Returns reference to current thread Vars_Manager
*
* @return Vars_Manager
*/
public function &get_current_thread_vm() {
return $this->threads[$this->get_current_thread_index()];
}
/**
* Checks if URL variable is passed in POST or GET
*
* @param Web_Context $context
* @return boolean
* @internal
*/
private function is_url_rewrite_var_passed(Web_Context $context) {
return $context->exists_in_get(THREADS_MANAGER_NAME)
|| $context->exists_in_post(THREADS_MANAGER_NAME);
}
/**
* Returns value of passed URL rewrite variable
*
* @param Web_Context $context
* @return string
* @internal
*/
private function get_url_rewrite_var(Web_Context $context) {
if ($context->exists_in_get(THREADS_MANAGER_NAME, $context)) {
$incoming_url_var = $context->get_from_get(THREADS_MANAGER_NAME);
} else {
$incoming_url_var = $context->get_from_post(THREADS_MANAGER_NAME);
}
return $incoming_url_var;
}
/**
* Adds new thread
*
* @return integer Index of the new thread
* @internal
*/
private function add_new_thread() {
$c = $this->next_thread_index;
$this->threads[$c] = new Thread_Vars_Manager();
$this->next_thread_index++;
return $c;
}
/**
* Generates random number
*
* @return integer
* @internal
*/
private function generate_new_random() {
return rand(0, 100000);
}
/**
* Generates new value for URL rewrite variable
*
* @return string
* @internal
*/
private function get_new_url_var() {
do {
$url_var = md5($this->generate_new_random());
} while(array_key_exists($url_var, $this->url_var_to_thread_links));
return $url_var;
}
/**
* Colects garbage snapshots
* @internal
*/
private function collect_garbage() {
foreach($this->threads_snapshots as $thread_index => $snapshot_arr) {
foreach($snapshot_arr as $snapshot_index => $s) {
if ($s['last_access_time'] + $this->snapshots_lifetime < time()) {
//printbr("Unsetting: $thread_index $snapshot_index");
unset($this->threads_snapshots[$thread_index][$snapshot_index]);
if (count($this->threads_snapshots[$thread_index]) == 0) {
unset($this->threads_snapshots[$thread_index]);
unset($this->threads[$thread_index]);
unset($this->next_snapshot_index[$thread_index]);
}
$this->remove_url_var_to_thread_link($thread_index, $snapshot_index);
} else {
//printbr('Thread: '.$thread_index.' Snapshot: '.$snapshot_index.' '.date("Y-m-d H:i:s", $s['last_access_time'] + $this->snapshots_lifetime) . ' - ' . date("Y-m-d H:i:s", time()));
}
}
}
}
/**
* Removes url_var_to_thread_link for given snapshot
*
* @param integer $thread_index
* @param integer $snapshot_index
* @internal
*/
private function remove_url_var_to_thread_link($thread_index, $snapshot_index) {
foreach($this->url_var_to_thread_links as $key => $link) {
if ($link['thread'] == $thread_index && $link['snapshot'] == $snapshot_index) {
//printbr('Removing link: '.$key.' T:'.$thread_index.' S:'.$snapshot_index);
unset($this->url_var_to_thread_links[$key]);
break;
}
}
}
/**
* Returns statistic about usage of threads and snapshots.
*
* @return unknown
*/
public function get_usage_stat() {
$ret['threads'] = count($this->threads);
$ret['snapshots'] = 0;
foreach($this->threads_snapshots as $ts) {
$ret['snapshots'] += count($ts);
}
return $ret;
}
public function reset() {
$this->threads_snapshots = array();
$this->threads = array();
$this->next_snapshot_index = array();
}
}

Source for file threads_manager.class.php