Page MenuHomestyx hydra
Files F1370

No OneTemporary
Actions

View Options

diff --git a/src/infrastructure/storage/connection/AphrontDatabaseConnection.php b/src/infrastructure/storage/connection/AphrontDatabaseConnection.php
index 0d4597b34f..f0eb2f9fe1 100644
--- a/src/infrastructure/storage/connection/AphrontDatabaseConnection.php
+++ b/src/infrastructure/storage/connection/AphrontDatabaseConnection.php
@@ -1,305 +1,301 @@
<?php
/**
* @task xaction Transaction Management
*/
abstract class AphrontDatabaseConnection
extends Phobject
implements PhutilQsprintfInterface {
private $transactionState;
private $readOnly;
private $queryTimeout;
private $locks = array();
private $lastActiveEpoch;
private $persistent;
abstract public function getInsertID();
abstract public function getAffectedRows();
abstract public function selectAllResults();
abstract public function executeQuery(PhutilQueryString $query);
abstract public function executeRawQueries(array $raw_queries);
abstract public function close();
abstract public function openConnection();
public function __destruct() {
// NOTE: This does not actually close persistent connections: PHP maintains
// them in the connection pool.
$this->close();
}
final public function setLastActiveEpoch($epoch) {
$this->lastActiveEpoch = $epoch;
return $this;
}
final public function getLastActiveEpoch() {
return $this->lastActiveEpoch;
}
final public function setPersistent($persistent) {
$this->persistent = $persistent;
return $this;
}
final public function getPersistent() {
return $this->persistent;
}
public function queryData($pattern/* , $arg, $arg, ... */) {
$args = func_get_args();
array_unshift($args, $this);
return call_user_func_array('queryfx_all', $args);
}
public function query($pattern/* , $arg, $arg, ... */) {
$args = func_get_args();
array_unshift($args, $this);
return call_user_func_array('queryfx', $args);
}
public function supportsAsyncQueries() {
return false;
}
- public function supportsParallelQueries() {
- return false;
- }
-
public function setReadOnly($read_only) {
$this->readOnly = $read_only;
return $this;
}
public function getReadOnly() {
return $this->readOnly;
}
public function setQueryTimeout($query_timeout) {
$this->queryTimeout = $query_timeout;
return $this;
}
public function getQueryTimeout() {
return $this->queryTimeout;
}
public function asyncQuery($raw_query) {
throw new Exception(pht('Async queries are not supported.'));
}
public static function resolveAsyncQueries(array $conns, array $asyncs) {
throw new Exception(pht('Async queries are not supported.'));
}
/**
* Is this connection idle and safe to close?
*
* A connection is "idle" if it can be safely closed without loss of state.
* Connections inside a transaction or holding locks are not idle, even
* though they may not actively be executing queries.
*
* @return bool True if the connection is idle and can be safely closed.
*/
public function isIdle() {
if ($this->isInsideTransaction()) {
return false;
}
if ($this->isHoldingAnyLock()) {
return false;
}
return true;
}
/* -( Global Locks )------------------------------------------------------- */
public function rememberLock($lock) {
if (isset($this->locks[$lock])) {
throw new Exception(
pht(
'Trying to remember lock "%s", but this lock has already been '.
'remembered.',
$lock));
}
$this->locks[$lock] = true;
return $this;
}
public function forgetLock($lock) {
if (empty($this->locks[$lock])) {
throw new Exception(
pht(
'Trying to forget lock "%s", but this connection does not remember '.
'that lock.',
$lock));
}
unset($this->locks[$lock]);
return $this;
}
public function forgetAllLocks() {
$this->locks = array();
return $this;
}
public function isHoldingAnyLock() {
return (bool)$this->locks;
}
/* -( Transaction Management )--------------------------------------------- */
/**
* Begin a transaction, or set a savepoint if the connection is already
* transactional.
*
* @return $this
* @task xaction
*/
public function openTransaction() {
$state = $this->getTransactionState();
$point = $state->getSavepointName();
$depth = $state->getDepth();
$new_transaction = ($depth == 0);
if ($new_transaction) {
$this->query('START TRANSACTION');
} else {
$this->query('SAVEPOINT '.$point);
}
$state->increaseDepth();
return $this;
}
/**
* Commit a transaction, or stage a savepoint for commit once the entire
* transaction completes if inside a transaction stack.
*
* @return $this
* @task xaction
*/
public function saveTransaction() {
$state = $this->getTransactionState();
$depth = $state->decreaseDepth();
if ($depth == 0) {
$this->query('COMMIT');
}
return $this;
}
/**
* Rollback a transaction, or unstage the last savepoint if inside a
* transaction stack.
*
* @return $this
*/
public function killTransaction() {
$state = $this->getTransactionState();
$depth = $state->decreaseDepth();
if ($depth == 0) {
$this->query('ROLLBACK');
} else {
$this->query('ROLLBACK TO SAVEPOINT '.$state->getSavepointName());
}
return $this;
}
/**
* Returns true if the connection is transactional.
*
* @return bool True if the connection is currently transactional.
* @task xaction
*/
public function isInsideTransaction() {
$state = $this->getTransactionState();
return ($state->getDepth() > 0);
}
/**
* Get the current @{class:AphrontDatabaseTransactionState} object, or create
* one if none exists.
*
* @return AphrontDatabaseTransactionState Current transaction state.
* @task xaction
*/
protected function getTransactionState() {
if (!$this->transactionState) {
$this->transactionState = new AphrontDatabaseTransactionState();
}
return $this->transactionState;
}
/**
* @task xaction
*/
public function beginReadLocking() {
$this->getTransactionState()->beginReadLocking();
return $this;
}
/**
* @task xaction
*/
public function endReadLocking() {
$this->getTransactionState()->endReadLocking();
return $this;
}
/**
* @task xaction
*/
public function isReadLocking() {
return $this->getTransactionState()->isReadLocking();
}
/**
* @task xaction
*/
public function beginWriteLocking() {
$this->getTransactionState()->beginWriteLocking();
return $this;
}
/**
* @task xaction
*/
public function endWriteLocking() {
$this->getTransactionState()->endWriteLocking();
return $this;
}
/**
* @task xaction
*/
public function isWriteLocking() {
return $this->getTransactionState()->isWriteLocking();
}
}
diff --git a/src/infrastructure/storage/connection/mysql/AphrontMySQLDatabaseConnection.php b/src/infrastructure/storage/connection/mysql/AphrontMySQLDatabaseConnection.php
index 40a2c6c357..c761fadfbb 100644
--- a/src/infrastructure/storage/connection/mysql/AphrontMySQLDatabaseConnection.php
+++ b/src/infrastructure/storage/connection/mysql/AphrontMySQLDatabaseConnection.php
@@ -1,233 +1,160 @@
<?php
final class AphrontMySQLDatabaseConnection
extends AphrontBaseMySQLDatabaseConnection {
public function escapeUTF8String($string) {
$this->validateUTF8String($string);
return $this->escapeBinaryString($string);
}
public function escapeBinaryString($string) {
return mysql_real_escape_string($string, $this->requireConnection());
}
public function getInsertID() {
return mysql_insert_id($this->requireConnection());
}
public function getAffectedRows() {
return mysql_affected_rows($this->requireConnection());
}
protected function closeConnection() {
mysql_close($this->requireConnection());
}
protected function connect() {
if (!function_exists('mysql_connect')) {
// We have to '@' the actual call since it can spew all sorts of silly
// noise, but it will also silence fatals caused by not having MySQL
// installed, which has bitten me on three separate occasions. Make sure
// such failures are explicit and loud.
throw new Exception(
pht(
'About to call %s, but the PHP MySQL extension is not available!',
'mysql_connect()'));
}
$user = $this->getConfiguration('user');
$host = $this->getConfiguration('host');
$port = $this->getConfiguration('port');
if ($port) {
$host .= ':'.$port;
}
$database = $this->getConfiguration('database');
$pass = $this->getConfiguration('pass');
if ($pass instanceof PhutilOpaqueEnvelope) {
$pass = $pass->openEnvelope();
}
$timeout = $this->getConfiguration('timeout');
$timeout_ini = 'mysql.connect_timeout';
if ($timeout) {
$old_timeout = ini_get($timeout_ini);
ini_set($timeout_ini, $timeout);
}
try {
$conn = @mysql_connect(
$host,
$user,
$pass,
$new_link = true,
$flags = 0);
} catch (Exception $ex) {
if ($timeout) {
ini_set($timeout_ini, $old_timeout);
}
throw $ex;
}
if ($timeout) {
ini_set($timeout_ini, $old_timeout);
}
if (!$conn) {
$errno = mysql_errno();
$error = mysql_error();
$this->throwConnectionException($errno, $error, $user, $host);
}
if ($database !== null) {
$ret = @mysql_select_db($database, $conn);
if (!$ret) {
$this->throwQueryException($conn);
}
}
$ok = @mysql_set_charset('utf8mb4', $conn);
if (!$ok) {
mysql_set_charset('binary', $conn);
}
return $conn;
}
protected function rawQuery($raw_query) {
return @mysql_query($raw_query, $this->requireConnection());
}
/**
* @phutil-external-symbol function mysql_multi_query
* @phutil-external-symbol function mysql_fetch_result
* @phutil-external-symbol function mysql_more_results
* @phutil-external-symbol function mysql_next_result
*/
protected function rawQueries(array $raw_queries) {
$conn = $this->requireConnection();
$results = array();
if (!function_exists('mysql_multi_query')) {
foreach ($raw_queries as $key => $raw_query) {
$results[$key] = $this->processResult($this->rawQuery($raw_query));
}
return $results;
}
if (!mysql_multi_query(implode("\n;\n\n", $raw_queries), $conn)) {
$ex = $this->processResult(false);
return array_fill_keys(array_keys($raw_queries), $ex);
}
$processed_all = false;
foreach ($raw_queries as $key => $raw_query) {
$results[$key] = $this->processResult(@mysql_fetch_result($conn));
if (!mysql_more_results($conn)) {
$processed_all = true;
break;
}
mysql_next_result($conn);
}
if (!$processed_all) {
throw new Exception(
pht('There are some results left in the result set.'));
}
return $results;
}
protected function freeResult($result) {
mysql_free_result($result);
}
- public function supportsParallelQueries() {
- // fb_parallel_query() doesn't support results with different columns.
- return false;
- }
-
- /**
- * @phutil-external-symbol function fb_parallel_query
- */
- public function executeParallelQueries(
- array $queries,
- array $conns = array()) {
- assert_instances_of($conns, __CLASS__);
-
- $map = array();
- $is_write = false;
- foreach ($queries as $id => $query) {
- $is_write = $is_write || $this->checkWrite($query);
- $conn = idx($conns, $id, $this);
-
- $host = $conn->getConfiguration('host');
- $port = 0;
- $match = null;
- if (preg_match('/(.+):(.+)/', $host, $match)) {
- list(, $host, $port) = $match;
- }
-
- $pass = $conn->getConfiguration('pass');
- if ($pass instanceof PhutilOpaqueEnvelope) {
- $pass = $pass->openEnvelope();
- }
-
- $map[$id] = array(
- 'sql' => $query,
- 'ip' => $host,
- 'port' => $port,
- 'username' => $conn->getConfiguration('user'),
- 'password' => $pass,
- 'db' => $conn->getConfiguration('database'),
- );
- }
-
- $profiler = PhutilServiceProfiler::getInstance();
- $call_id = $profiler->beginServiceCall(
- array(
- 'type' => 'multi-query',
- 'queries' => $queries,
- 'write' => $is_write,
- ));
-
- $map = fb_parallel_query($map);
-
- $profiler->endServiceCall($call_id, array());
-
- $results = array();
- $pos = 0;
- $err_pos = 0;
- foreach ($queries as $id => $query) {
- $errno = idx(idx($map, 'errno', array()), $err_pos);
- $err_pos++;
- if ($errno) {
- try {
- $this->throwQueryCodeException($errno, $map['error'][$id]);
- } catch (Exception $ex) {
- $results[$id] = $ex;
- }
- continue;
- }
- $results[$id] = $map['result'][$pos];
- $pos++;
- }
- return $results;
- }
-
protected function fetchAssoc($result) {
return mysql_fetch_assoc($result);
}
protected function getErrorCode($connection) {
return mysql_errno($connection);
}
protected function getErrorDescription($connection) {
return mysql_error($connection);
}
}
diff --git a/src/infrastructure/storage/future/QueryFuture.php b/src/infrastructure/storage/future/QueryFuture.php
index 0f1fc3b9d4..252a136e73 100644
--- a/src/infrastructure/storage/future/QueryFuture.php
+++ b/src/infrastructure/storage/future/QueryFuture.php
@@ -1,129 +1,117 @@
<?php
/**
* This class provides several approaches for querying data from the database:
*
* # Async queries: Used under MySQLi with MySQLnd.
* # Parallel queries: Used under HPHP.
* # Multi queries: Used under MySQLi or HPHP.
* # Single queries: Used under MySQL.
*
* The class automatically decides which approach to use. Usage is like with
* other futures:
*
* $futures = array();
* $futures[] = new QueryFuture($conn1, 'SELECT 1');
* $futures[] = new QueryFuture($conn1, 'DELETE FROM table');
* $futures[] = new QueryFuture($conn2, 'SELECT 2');
*
* foreach (new FutureIterator($futures) as $future) {
* try {
* $result = $future->resolve();
* } catch (AphrontQueryException $ex) {
* }
* }
*
* `$result` contains a list of dicts for select queries or number of modified
* rows for modification queries.
*/
final class QueryFuture extends Future {
private static $futures = array();
private $conn;
private $query;
private $id;
private $async;
private $profilerCallID;
public function __construct(
AphrontDatabaseConnection $conn,
$pattern/* , ... */) {
$this->conn = $conn;
$args = func_get_args();
$args = array_slice($args, 2);
$this->query = vqsprintf($conn, $pattern, $args);
self::$futures[] = $this;
$this->id = last_key(self::$futures);
}
public function isReady() {
if ($this->canResolve()) {
return true;
}
if (!$this->conn->supportsAsyncQueries()) {
- if ($this->conn->supportsParallelQueries()) {
- $queries = array();
- $conns = array();
- foreach (self::$futures as $id => $future) {
- $queries[$id] = $future->query;
- $conns[$id] = $future->conn;
- }
- $results = $this->conn->executeParallelQueries($queries, $conns);
- $this->processResults($results);
- return true;
- }
-
$conns = array();
$conn_queries = array();
foreach (self::$futures as $id => $future) {
$hash = spl_object_hash($future->conn);
$conns[$hash] = $future->conn;
$conn_queries[$hash][$id] = $future->query;
}
foreach ($conn_queries as $hash => $queries) {
$this->processResults($conns[$hash]->executeRawQueries($queries));
}
return true;
}
if (!$this->async) {
$profiler = PhutilServiceProfiler::getInstance();
$this->profilerCallID = $profiler->beginServiceCall(
array(
'type' => 'query',
'query' => $this->query,
'async' => true,
));
$this->async = $this->conn->asyncQuery($this->query);
return false;
}
$conns = array();
$asyncs = array();
foreach (self::$futures as $id => $future) {
if ($future->async) {
$conns[$id] = $future->conn;
$asyncs[$id] = $future->async;
}
}
$this->processResults($this->conn->resolveAsyncQueries($conns, $asyncs));
if ($this->canResolve()) {
return true;
}
return false;
}
private function processResults(array $results) {
foreach ($results as $id => $result) {
$future = self::$futures[$id];
if ($result instanceof Exception) {
$future->exception = $result;
} else {
$future->result = $result;
}
unset(self::$futures[$id]);
if ($future->profilerCallID) {
$profiler = PhutilServiceProfiler::getInstance();
$profiler->endServiceCall($future->profilerCallID, array());
}
}
}
}

File Metadata

Mime Type
text/x-diff
Expires
Mon, Nov 25, 1:11 AM (1 d, 15 h)
Storage Engine
blob
Storage Format
Raw Data
Storage Handle
1138
Default Alt Text
(17 KB)

Event Timeline

styx copyright 2020 sirocyl / ellie (kymkdd) / ncommander / styx contributors
Use of other names and brands is not intended to reflect a claim of, or right to, use the name or brand.
styx is an operating system using the Linux kernel and components, and "styx Linux" as a phrase does not reflect the inclusion of the Linux brand or trademark in the styx project.
Linux® is the registered trademark of Linus Torvalds in the U.S. and other countries.
Terms and Conditions (TODO: fix this!)