<?php // This is called by cron, and is persistent. // It takes care of fast caching, like memcache. // It will receive signals using a UNIX socket. error_reporting(E_ALL); ini_set("track_errors", "On"); set_time_limit(0); // Increment this if we need to restart this daemon (protocol changes etc.) $rg_cache_version = 36; $_s = microtime(TRUE); require_once("/etc/rocketgit/config.php"); $INC = dirname(__FILE__) . "/../inc"; require_once($INC . "/init.inc.php"); require_once($INC . "/log.inc.php"); require_once($INC . "/sql.inc.php"); require_once($INC . "/struct.inc.php"); require_once($INC . "/cache.inc.php"); require_once($INC . "/repo.inc.php"); require_once($INC . "/prof.inc.php"); require_once($INC . "/mr.inc.php"); require_once($INC . "/keys.inc.php"); require_once($INC . "/user.inc.php"); require_once($INC . "/bug.inc.php"); require_once($INC . "/fixes.inc.php"); require_once($INC . "/ver.php"); function rg_destroy($k, &$conn_table) { rg_log("Destroying key $k..."); if (isset($conn_table['r'][$k])) unset($conn_table['r'][$k]); if (isset($conn_table['w'][$k])) unset($conn_table['w'][$k]); // TODO: seems socket is already closed if (isset($conn_table['conns'][$k]['socket'])) if (is_resource($conn_table['conns'][$k]['socket'])) socket_close($conn_table['conns'][$k]['socket']); unset($conn_table['conns'][$k]); } function rg_handle_command($k, &$conn_table, $cmd) { //rg_log("rg_handle_command: k=$k, cmd=$cmd"); $s = &$conn_table['conns'][$k]; $a = explode(' ', $cmd, 4); $buf = "ER Invalid command\n"; $no_wait = FALSE; while (1) { // We must have at least 2 parameters: cmd and flags and I if (!isset($a[2])) break; $cmd = trim($a[0]); $flags = trim($a[1]); if (strncmp($flags, 'F=', 2) != 0) { rg_log('Invalid command (no flags): $cmd'); break; } $flags = substr($flags, 2); if (strstr($flags, 'W')) $no_wait = TRUE; $id = trim($a[2]); if (strncmp($id, 'I=', 2) != 0) { rg_log('Invalid command (no id): $cmd'); break; } $id = substr($id, 2); /* From here, commands with no parameters */ /* none yet */ /* From here, at least 1 para */ if (!isset($a[3])) break; $para1 = trim($a[3]); if (strcmp($cmd, "SET") == 0) { $ns_var_value = explode("=", $para1, 2); if (!isset($ns_var_value[1])) break; $ns_var = trim($ns_var_value[0]); $value = trim($ns_var_value[1]); $value = unserialize(stripcslashes($value)); if ($value !== FALSE) { rg_cache_core_set("normal::" . $ns_var, $value); $buf = 'OK ' . $id . "\n"; } else { $buf = 'ER ' . $id . ' cannot unserialize data' . "\n"; } break; } if (strcmp($cmd, "MERGE") == 0) { $ns_var_value = explode("=", $para1, 2); if (!isset($ns_var_value[1])) break; $ns_var = trim($ns_var_value[0]); $value = trim($ns_var_value[1]); $value = unserialize(stripcslashes($value)); if ($value !== FALSE) { $ret = rg_cache_core_merge("normal::" . $ns_var, $value); if ($ret === FALSE) $buf = 'ER ' . $id . ' NOT_FOUND' . "\n"; else $buf = 'OK ' . $id . "\n"; } else { $buf = 'ER ' . $id . 'cannot unserialize data' . "\n"; } break; } if (strcmp($cmd, "INC") == 0) { $v = rg_cache_core_inc("normal::" . $para1); $buf = 'OK ' . $id . ' v=' . $v . "\n"; break; } if (strcmp($cmd, "GET") == 0) { $ret = rg_cache_core_get("normal::" . $para1); if ($ret === FALSE) $buf = 'ER ' . $id . ' NOT_FOUND' . "\n"; else $buf = 'OK ' . $id . ' ' . rg_cache_prepare($ret) . "\n"; break; } if (strcmp($cmd, "UNSET") == 0) { $ret = rg_cache_core_unset("normal::" . $para1); if ($ret === FALSE) $buf = 'ER ' . $id . ' NOT_FOUND' . "\n"; else $buf = 'OK ' . $id . "\n"; break; } if (strcmp($cmd, "ADUMP") == 0) { $ret = rg_cache_core_adump("normal::" . $para1); if ($ret === FALSE) $buf = 'ER ' . $id . ' NOT_FOUND' . "\n"; else $buf = 'OK ' . $id . ' ' . rg_cache_prepare($ret) . "\n"; break; } if (strcmp($cmd, "APUSH") == 0) { $ns_var_value = explode("=", $para1, 2); if (!isset($ns_var_value[1])) break; $ns_var = trim($ns_var_value[0]); $value = trim($ns_var_value[1]); $value = unserialize(stripcslashes($value)); if ($value !== FALSE) { rg_cache_core_apush("normal::" . $ns_var, $value); $buf = 'OK ' . $id . "\n"; } else { $buf = 'ER ' . $id . ' cannot unserialize data' . "\n"; } break; } if (strcmp($cmd, "APOP") == 0) { $ret = rg_cache_core_apop("normal::" . $para1); if ($ret === FALSE) $buf = 'ER ' . $id . ' NOT_FOUND' . "\n"; else $buf = 'OK ' . $id . ' ' . $ret . "\n"; break; } if (strcmp($cmd, "ASHIFT") == 0) { $ret = rg_cache_core_ashift("normal::" . $para1); if ($ret === FALSE) $buf = 'ER ' . $id . ' NOT_FOUND' . "\n"; else $buf = 'OK ' . $id . ' ' . $ret . "\n"; break; } if (strcmp($cmd, "SHUTDOWN") == 0) { rg_log('Shutting down...'); exit(0); } if (strcmp($cmd, "SLEEP") == 0) { rg_log('Sleeping...'); $buf = 'OK ' . $id . "\n"; sleep(3); } break; } if ($no_wait === FALSE) { rg_log($k . ':DEBUG: enqueue: ' . $buf); $s['send'] .= $buf; $conn_table['w'][$k] = $s['socket']; } } function rg_handle_recv($k, &$conn_table) { //rg_log("handle_recv on key $k..."); $s = &$conn_table['conns'][$k]; $ret = @socket_recv($s['socket'], $buf, 32 * 4096, 0); if ($ret === FALSE) { rg_log("Error in recv (" . socket_strerror(socket_last_error()) . ")"); rg_destroy($k, $conn_table); return; } if ($ret === 0) { rg_log("Remote closed the connection (received 0)."); rg_destroy($k, $conn_table); return; } rg_log($k . ':DEBUG: received: ' . $buf); $s['recv'] .= $buf; $s['close_at'] = time() + 30; while (1) { $pos = strpos($s['recv'], "\n"); if ($pos === FALSE) return; $cmd = substr($s['recv'], 0, $pos); rg_handle_command($k, $conn_table, $cmd); $s['recv'] = substr($s['recv'], $pos + 1); } } function rg_handle_send($k, &$conn_table) { //rg_log("Sending on key $k..."); $s = &$conn_table['conns'][$k]; $ret = @socket_send($s['socket'], $s['send'], strlen($s['send']), 0); if ($ret === FALSE) { rg_log("Cannot send (" . socket_strerror(socket_last_error()) . ")"); rg_destroy($k, $conn_table); return; } $s['send'] = substr($s['send'], $ret); if (empty($s['send'])) unset($conn_table['w'][$k]); } function rg_handle_new($client, &$conn_table) { socket_set_nonblock($client); $key = intval($client); $conn_table['conns'][$key] = array( "socket" => $client, "recv" => "", "send" => "", "close_at" => time() + 30 ); $conn_table['r'][$key] = $client; //rg_log("Added client with key $key."); /* This way I can enforce the connecting user to be apache/rocketgit but is seems it does not work correctly in PHP... $t = socket_get_option($client, SOL_SOCKET, 17 //SO_PEERCRED); rg_log_ml('PEERCRED: ' . print_r($t, TRUE)); */ } function rg_handle_idle(&$conn_table) { $now = time(); foreach ($conn_table['conns'] as $key => $info) { if (!isset($info['close_at'])) { rg_internal_error('close_at is not set!'); continue; } if ($info['close_at'] < $now) { rg_log("Destroy $key because has been too much time idle."); rg_destroy($key, $conn_table); } } } rg_prof_start("MAIN"); rg_log_set_file($rg_log_dir . "/cache.log"); rg_log_set_sid("000000"); // to spread the logs rg_log("Start (ver=$rocketgit_version)..."); // Remove the socket, else we will get error if (file_exists($rg_cache_socket)) unlink($rg_cache_socket); $master = socket_create(AF_UNIX, SOCK_STREAM, 0); if ($master === FALSE) { rg_internal_error("Cannot create events socket!"); exit(1); } $r = socket_bind($master, $rg_cache_socket); if ($r === FALSE) { rg_internal_error("Cannot bind socket!"); exit(1); } $r = socket_listen($master, 128); if ($r === FALSE) { rg_internal_error("Cannot set queue length on socket!"); exit(1); } socket_set_nonblock($master); // Allow apache (at least) to connect to socket // TODO: this is a security hole. Take care. Hm. How?! $r = chmod($rg_cache_socket, 0666); if ($r === FALSE) { rg_internal_error("Cannot set rights on cache socket!"); exit(1); } $conn_table = array("r" => array(), "w" => array(), "conns" => array()); $conn_table['r']['master'] = $master; do { rg_log_buffer_clear(); //rg_log_ml("conn_table: " . print_r($conn_table, TRUE)); $r2 = $conn_table['r']; $w2 = $conn_table['w']; $ex = array(); $r = @socket_select($r2, $w2, $ex, 5); if ($r === FALSE) rg_fatal("Cannot select (" . socket_strerror(socket_last_error()) . ")!"); if ($r > 0) { //if (!empty($r2)) // rg_log_ml('r2: ' . print_r($r2, TRUE)); //if (!empty($w2)) // rg_log_ml('w2: ' . print_r($w2, TRUE)); //if (!empty($ex)) // rg_log_ml('ex: ' . print_r($ex, TRUE)); foreach ($r2 as $key => $socket) { if (strcmp($key, "master") == 0) { $client = @socket_accept($socket); if ($client === FALSE) { rg_log("Connection seems broken!"); continue; } rg_handle_new($client, $conn_table); continue; } rg_handle_recv($key, $conn_table); } foreach ($w2 as $key => $sock) { if (isset($conn_table['conns'][$key])) rg_handle_send($key, $conn_table); } foreach ($ex as $key => $sock) rg_destroy($key, $conn_table); } // TODO: if we activate this, we broke the connection with events.php //rg_handle_idle($conn_table); } while (1); socket_close($master); rg_log("Exiting..."); rg_prof_end("MAIN"); rg_prof_log(); ?>