Disable isset, optimize getInstance, concurrent offsetSet
This commit is contained in:
parent
ca03bc662a
commit
321787f718
|
@ -8,10 +8,21 @@ use Amp\Promise;
|
||||||
interface DbArray extends DbType, \ArrayAccess, \Countable
|
interface DbArray extends DbType, \ArrayAccess, \Countable
|
||||||
{
|
{
|
||||||
public function getArrayCopy(): Promise;
|
public function getArrayCopy(): Promise;
|
||||||
public function offsetExists($offset): Promise;
|
public function isset($key): Promise;
|
||||||
public function offsetGet($offset): Promise;
|
public function offsetGet($offset): Promise;
|
||||||
public function offsetSet($offset, $value);
|
public function offsetSet($offset, $value);
|
||||||
public function offsetUnset($offset): Promise;
|
public function offsetUnset($offset): Promise;
|
||||||
public function count(): Promise;
|
public function count(): Promise;
|
||||||
public function getIterator(): Producer;
|
public function getIterator(): Producer;
|
||||||
|
|
||||||
|
/**
|
||||||
|
* @deprecated
|
||||||
|
* @internal
|
||||||
|
* @see DbArray::isset();
|
||||||
|
*
|
||||||
|
* @param mixed $offset
|
||||||
|
*
|
||||||
|
* @return bool
|
||||||
|
*/
|
||||||
|
public function offsetExists($offset);
|
||||||
}
|
}
|
|
@ -28,9 +28,14 @@ class MemoryArray extends \ArrayIterator implements DbArray
|
||||||
});
|
});
|
||||||
}
|
}
|
||||||
|
|
||||||
public function offsetExists($offset): Promise
|
public function offsetExists($offset)
|
||||||
{
|
{
|
||||||
return call(fn() => parent::offsetExists($offset));
|
throw new \RuntimeException('Native isset not support promises. Use isset method');
|
||||||
|
}
|
||||||
|
|
||||||
|
public function isset($key): Promise
|
||||||
|
{
|
||||||
|
return call(fn() => parent::offsetExists($key));
|
||||||
}
|
}
|
||||||
|
|
||||||
public function offsetGet($offset): Promise
|
public function offsetGet($offset): Promise
|
||||||
|
|
|
@ -2,14 +2,12 @@
|
||||||
|
|
||||||
namespace danog\MadelineProto\Db;
|
namespace danog\MadelineProto\Db;
|
||||||
|
|
||||||
use Amp\Loop;
|
|
||||||
use Amp\Mysql\Pool;
|
use Amp\Mysql\Pool;
|
||||||
use Amp\Producer;
|
use Amp\Producer;
|
||||||
use Amp\Promise;
|
use Amp\Promise;
|
||||||
use Amp\Sql\ResultSet;
|
use Amp\Sql\ResultSet;
|
||||||
use danog\MadelineProto\Logger;
|
use danog\MadelineProto\Logger;
|
||||||
use function Amp\call;
|
use function Amp\call;
|
||||||
use function Amp\Promise\wait;
|
|
||||||
|
|
||||||
class MysqlArray implements DbArray
|
class MysqlArray implements DbArray
|
||||||
{
|
{
|
||||||
|
@ -50,9 +48,14 @@ class MysqlArray implements DbArray
|
||||||
*/
|
*/
|
||||||
public static function getInstance(string $name, $value = null, string $tablePrefix = '', array $settings = []): Promise
|
public static function getInstance(string $name, $value = null, string $tablePrefix = '', array $settings = []): Promise
|
||||||
{
|
{
|
||||||
$instance = new static();
|
$tableName = "{$tablePrefix}_{$name}";
|
||||||
|
if ($value instanceof self && $value->table === $tableName) {
|
||||||
|
$instance = &$value;
|
||||||
|
} else {
|
||||||
|
$instance = new static();
|
||||||
|
$instance->table = $tableName;
|
||||||
|
}
|
||||||
|
|
||||||
$instance->table = "{$tablePrefix}_{$name}";
|
|
||||||
$instance->settings = $settings;
|
$instance->settings = $settings;
|
||||||
$instance->db = static::getDbConnection($settings);
|
$instance->db = static::getDbConnection($settings);
|
||||||
$instance->ttl = $settings['cache_ttl'] ?? $instance->ttl;
|
$instance->ttl = $settings['cache_ttl'] ?? $instance->ttl;
|
||||||
|
@ -60,9 +63,13 @@ class MysqlArray implements DbArray
|
||||||
$instance->startCacheCleanupLoop();
|
$instance->startCacheCleanupLoop();
|
||||||
|
|
||||||
return call(static function() use($instance, $value) {
|
return call(static function() use($instance, $value) {
|
||||||
yield from static::renameTmpTable($instance, $value);
|
|
||||||
yield from $instance->prepareTable();
|
yield from $instance->prepareTable();
|
||||||
yield from static::migrateDataToDb($instance, $value);
|
|
||||||
|
//Skip migrations if its same object
|
||||||
|
if ($instance !== $value) {
|
||||||
|
yield from static::renameTmpTable($instance, $value);
|
||||||
|
yield from static::migrateDataToDb($instance, $value);
|
||||||
|
}
|
||||||
|
|
||||||
return $instance;
|
return $instance;
|
||||||
});
|
});
|
||||||
|
@ -109,7 +116,7 @@ class MysqlArray implements DbArray
|
||||||
$total = count($value);
|
$total = count($value);
|
||||||
foreach ($value as $key => $item) {
|
foreach ($value as $key => $item) {
|
||||||
$counter++;
|
$counter++;
|
||||||
if ($counter % 100 === 0) {
|
if ($counter % 500 === 0) {
|
||||||
yield $instance->offsetSet($key, $item);
|
yield $instance->offsetSet($key, $item);
|
||||||
Logger::log("Loading data to table {$instance->table}: $counter/$total", Logger::WARNING);
|
Logger::log("Loading data to table {$instance->table}: $counter/$total", Logger::WARNING);
|
||||||
} else {
|
} else {
|
||||||
|
@ -121,21 +128,21 @@ class MysqlArray implements DbArray
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
public function offsetExists($index): bool
|
||||||
|
{
|
||||||
|
throw new \RuntimeException('Native isset not support promises. Use isset method');
|
||||||
|
}
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* Check if offset exists
|
* Check if key isset
|
||||||
*
|
*
|
||||||
* @link https://php.net/manual/en/arrayiterator.offsetexists.php
|
* @param $key
|
||||||
*
|
|
||||||
* @param string $index <p>
|
|
||||||
* The offset being checked.
|
|
||||||
* </p>
|
|
||||||
*
|
*
|
||||||
* @return Promise<bool> true if the offset exists, otherwise false
|
* @return Promise<bool> true if the offset exists, otherwise false
|
||||||
* @throws \Throwable
|
|
||||||
*/
|
*/
|
||||||
public function offsetExists($index): Promise
|
public function isset($key): Promise
|
||||||
{
|
{
|
||||||
return call(fn() => yield $this->offsetGet($index) !== null);
|
return call(fn() => yield $this->offsetGet($key) !== null);
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
||||||
|
@ -177,18 +184,24 @@ class MysqlArray implements DbArray
|
||||||
if ($this->getCache($index) === $value) {
|
if ($this->getCache($index) === $value) {
|
||||||
return call(fn()=>null);
|
return call(fn()=>null);
|
||||||
}
|
}
|
||||||
|
|
||||||
$this->setCache($index, $value);
|
$this->setCache($index, $value);
|
||||||
|
|
||||||
return $this->request("
|
$request = $this->request("
|
||||||
INSERT INTO `{$this->table}`
|
INSERT INTO `{$this->table}`
|
||||||
SET `key` = :index, `value` = :value
|
SET `key` = :index, `value` = :value
|
||||||
ON DUPLICATE KEY UPDATE `value` = :value
|
ON DUPLICATE KEY UPDATE `value` = :value
|
||||||
",
|
",
|
||||||
[
|
[
|
||||||
'index' => $index,
|
'index' => $index,
|
||||||
'value' => serialize($value),
|
'value' => serialize($value),
|
||||||
]
|
]
|
||||||
);
|
);
|
||||||
|
|
||||||
|
//Ensure that cache is synced with latest insert in case of concurrent requests.
|
||||||
|
$request->onResolve(fn() => $this->setCache($index, $value));
|
||||||
|
|
||||||
|
return $request;
|
||||||
}
|
}
|
||||||
|
|
||||||
/**
|
/**
|
||||||
|
|
Loading…
Reference in New Issue
Block a user