その場でデータベースを変更する機能を備えた単純なORMを作成する

こんにちは、Habr！カルマは、ホリバーの記事の下での不注意なコメントのために排水されました。つまり、興味深い（私は願っています）投稿を書いて、自分自身をリハビリする必要があります。



私は数年間、phpでテレグラムサーバークライアントを使用しています。そして、多くのユーザーのように、メモリ消費量の絶え間ない増加にうんざりしています。一部のセッションでは、1〜8ギガバイトのRAMが必要になる場合があります。データベースのサポートは長い間約束されてきましたが、この方向への進展はありませんでした。私は自分で問題を解決しなければなりませんでした:)オープンソースプロジェクトの人気はプルリクエストに興味深い要件を課しました：

1. 後方互換性。既存のすべてのセッションは、新しいバージョンでも引き続き機能する必要があります（セッションは、ファイル内のアプリケーションのシリアル化されたインスタンスです）。
2. データベースの選択の自由。ユーザーは環境の構成が異なるため、データを失うことなく、いつでもストレージタイプを変更する機能。
3. 拡張性。新しいタイプのデータベースの追加のしやすさ。
4. インターフェイスを保存します。データを操作するアプリケーションコードは変更しないでください。
5. 非同期。プロジェクトはamphpを使用するため、すべてのデータベース操作は非ブロッキングである必要があります。

詳しくは猫の下のみんなを招待します。

何を転送しますか

MadelineProtoのメモリのほとんどは、チャット、ユーザー、およびファイルによって占められています。たとえば、ピアキャッシュには、2万を超えるエントリがあります。これらは、アカウントがこれまでに見たすべてのユーザー（すべてのグループのメンバーを含む）、およびチャネル、ボット、グループです。アカウントが古く、アクティブであるほど、より多くのデータがメモリに保存されます。これらは数十メガバイトと数百メガバイトであり、それらのほとんどは使用されていません。ただし、同じデータを複数回受信しようとすると、テレグラムによってアカウントがすぐに厳しく制限されるため、キャッシュ全体をクリアすることはできません。たとえば、パブリックデモサーバーでセッションを再作成した後、1週間以内の電報は、ほとんどのリクエストにFLOOD_WAITエラーで応答し、実際には何も機能しませんでした。キャッシュがウォームアップした後、すべてが正常に戻りました。



コードの観点から、このデータはクラスのペアのプロパティに配列として格納されます。

建築

要件に基づいて、スキームが生まれました。

• すべての「重い」配列は、ArrayAccessを実装するオブジェクトに置き換えられます。
• データベースの種類ごとに、基本クラスを継承する独自のクラスを作成します。
• オブジェクトは、__ consrtuctおよび__awake中に作成され、プロパティに書き込まれます。
• 抽象ファクトリは、アプリケーション設定で選択されたデータベースに応じて、オブジェクトに必要なクラスを選択します。
• アプリケーションにすでに別のタイプのストレージがある場合は、そこからすべてのデータを読み取り、アレイを新しいストレージに書き込みます。

非同期の世界の問題

私が最初にしたことは、メモリに配列を格納するためのインターフェイスとクラスを作成することでした。これはデフォルトであり、動作は古いバージョンのプログラムと同じです。最初の夜、私はプロトタイプの成功に非常に興奮しました。コードは素晴らしく、シンプルでした。これまでのところ、Iteratorインターフェースのメソッド内、およびunsetとissetを担当するメソッド内でジェネレーターを使用できないことは発見されていません。



ここで、amphpがジェネレーター構文を使用してphpに非同期を実装することを明確にする必要があります。歩留まりは非同期に類似するようになります... jsから待ちます。メソッドが非同期を使用する場合、そのメソッドから結果を取得するには、yieldを使用してコードでこの結果を待つ必要があります。例えば：

<?php

include 'vendor/autoload.php';

$MadelineProto = new \danog\MadelineProto\API('session.madeline');
$MadelineProto->async(true);

$MadelineProto->loop(function() use($MadelineProto) {
    $myAsyncFunction = function() use($MadelineProto): \Generator {
        $me = yield $MadelineProto->start();
        yield $MadelineProto->echo(json_encode($me, JSON_PRETTY_PRINT|JSON_UNESCAPED_UNICODE));
    };

    yield $myAsyncFunction();
});

文字列からの場合

yield $myAsyncFunction();

イールドを削除すると、このコードが実行される前にアプリケーションが終了します。結果は得られません。



メソッドや関数を呼び出す前にyieldを追加することはそれほど難しくありません。ただし、ArrayAccessインターフェイスが使用されるため、メソッドは直接呼び出されません。たとえば、unset（）はoffsetUnset（）を呼び出し、isset（）はoffsetIsset（）を呼び出します。 Iteratorインターフェースを使用する場合、foreachイテレーターの場合も状況は似ています。



組み込みメソッドの前にyieldを追加すると、これらのメソッドはジェネレーターで動作するように設計されていないため、エラーが発生します。コメントでもう少し：こことここ。



自分のメソッドを使用するには、コードを妥協して書き直す必要がありました。幸いなことに、そのような場所はほとんどありませんでした。ほとんどの場合、キーによる読み取りまたは書き込みには配列が使用されていました。この機能は、ジェネレーターと素晴らしい友達になりました。



結果のインターフェイスは次のとおりです。

<?php

use Amp\Producer;
use Amp\Promise;

interface DbArray extends DbType, \ArrayAccess, \Countable
{
    public function getArrayCopy(): Promise;
    public function isset($key): Promise;
    public function offsetGet($offset): Promise;
    public function offsetSet($offset, $value);
    public function offsetUnset($offset): Promise;
    public function count(): Promise;
    public function getIterator(): Producer;

    /**
     * @deprecated
     * @internal
     * @see DbArray::isset();
     *
     * @param mixed $offset
     *
     * @return bool
     */
    public function offsetExists($offset);
}

データの操作例

<?php
...
//
$existingChat = yield $this->chats[$user['id']];

//. 
yield $this->chats[$user['id']] = $user;
//   yield,           .
$this->chats[$user['id']] = $user;


//unset
yield $this->chats->offsetUnset($id);

//foreach
$iterator = $this->chats->getIterator();
while (yield $iterator->advance()) {
    [$key, $value] = $iterator->getCurrent();
    //  
}

データストレージ

データを保存する最も簡単な方法はシリアル化されています。オブジェクトをサポートするために、jsonの使用を断念しなければなりませんでした。このテーブルには、キーと値の2つの主要な列があります。



テーブルを作成するためのsqlクエリの例：

            CREATE TABLE IF NOT EXISTS `{$this->table}`
            (
                `key` VARCHAR(255) NOT NULL,
                `value` MEDIUMBLOB NULL,
                `ts` TIMESTAMP DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
                PRIMARY KEY (`key`)
            )
            ENGINE = InnoDB
            CHARACTER SET 'utf8mb4' 
            COLLATE 'utf8mb4_general_ci'

アプリケーションが起動するたびに、プロパティごとにテーブルを作成しようとします。テレグラムクライアントは、数時間に1回以上再起動することはお勧めしません。そのため、1秒あたりにテーブルを作成するための複数のリクエストはありません:)



プライマリキーは自動インクリメントされないため、データの挿入と更新は、通常の配列のように1つのクエリで実行できます。

INSERT INTO `{$this->table}` 
            SET `key` = :index, `value` = :value 
            ON DUPLICATE KEY UPDATE `value` = :value

％account_id％_％class％_％variable_name％の形式の名前を持つテーブルが変数ごとに作成されます。ただし、最初にアプリケーションを起動したときは、まだアカウントがありません。この場合、tmpプレフィックスを使用してランダムな一時IDを生成する必要があります。起動するたびに、各変数のクラスはアカウントIDが表示されているかどうかを確認します。idが存在する場合、テーブルの名前が変更されます。

インデックス

データベースの構造は可能な限り単純であるため、将来、新しいプロパティが自動的に追加されます。接続はありません。PRIMARYキーインデックスのみが使用されます。ただし、他のフィールドで検索する必要がある場合もあります。



たとえば、配列/テーブルチャットがあります。その中の鍵はチャットIDです。しかし、多くの場合、ユーザー名で検索する必要があります。アプリケーションがデータを配列に格納しているとき、ユーザー名による検索は、foreachで配列を反復することによって通常どおり実行されました。この検索は、メモリでは許容可能な速度で機能しましたが、データベースでは機能しませんでした。したがって、別のテーブル/配列が作成され、クラス内の対応するプロパティが作成されました。キーはユーザー名、値はチャットIDです。このアプローチの唯一の欠点は、2つのテーブルを同期するために追加のコードを作成する必要があることです。

キャッシング

ローカルmysqlは高速ですが、少しのキャッシュで問題が発生することはありません。特に、同じ値が連続して複数回使用される場合。たとえば、最初にデータベース内のチャットの存在を確認してから、そこからデータを取得します。



簡単な自転車の特性が書かれました。

<?php

namespace danog\MadelineProto\Db;

use Amp\Loop;
use danog\MadelineProto\Logger;

trait ArrayCacheTrait
{
    /**
     * Values stored in this format:
     * [
     *      [
     *          'value' => mixed,
     *          'ttl' => int
     *      ],
     *      ...
     * ].
     * @var array
     */
    protected array $cache = [];
    protected string $ttl = '+5 minutes';
    private string $ttlCheckInterval = '+1 minute';

    protected function getCache(string $key, $default = null)
    {
        $cacheItem = $this->cache[$key] ?? null;
        $result = $default;

        if (\is_array($cacheItem)) {
            $result = $cacheItem['value'];
            $this->cache[$key]['ttl'] = \strtotime($this->ttl);
        }

        return $result;
    }

    /**
     * Save item in cache.
     *
     * @param string $key
     * @param $value
     */
    protected function setCache(string $key, $value): void
    {
        $this->cache[$key] = [
            'value' => $value,
            'ttl' => \strtotime($this->ttl),
        ];
    }

    /**
     * Remove key from cache.
     *
     * @param string $key
     */
    protected function unsetCache(string $key): void
    {
        unset($this->cache[$key]);
    }

    protected function startCacheCleanupLoop(): void
    {
        Loop::repeat(\strtotime($this->ttlCheckInterval, 0) * 1000, fn () => $this->cleanupCache());
    }

    /**
     * Remove all keys from cache.
     */
    protected function cleanupCache(): void
    {
        $now = \time();
        $oldKeys = [];
        foreach ($this->cache as $cacheKey => $cacheValue) {
            if ($cacheValue['ttl'] < $now) {
                $oldKeys[] = $cacheKey;
            }
        }
        foreach ($oldKeys as $oldKey) {
            $this->unsetCache($oldKey);
        }

        Logger::log(
            \sprintf(
                "cache for table:%s; keys left: %s; keys removed: %s",
                $this->table,
                \count($this->cache),
                \count($oldKeys)
            ),
            Logger::VERBOSE
        );
    }
}

startCacheCleanupLoopに特に注意を払いたいと思います。 amphpの魔法のおかげで、キャッシュの無効化は可能な限り簡単です。コールバックは指定された間隔で開始し、すべての値をループして、この要素への最後の呼び出しのタイムスタンプを格納するtsフィールドを調べます。呼び出しが5分以上前（設定で構成可能）であった場合、要素は削除されます。 amphpを使用してredisまたはmemcacheからttlアナログを実装するのは非常に簡単です。これはすべてバックグラウンドで発生し、メインスレッドをブロックしません。



キャッシュと非同期性の助けを借りて、読み取りだけでなく書き込みも高速化されます。



データベースにデータを書き込むメソッドのソースコードは次のとおりです。

/**
     * Set value for an offset.
     *
     * @link https://php.net/manual/en/arrayiterator.offsetset.php
     *
     * @param string $index <p>
     * The index to set for.
     * </p>
     * @param $value
     *
     * @throws \Throwable
     */

    public function offsetSet($index, $value): Promise
    {
        if ($this->getCache($index) === $value) {
            return call(fn () =>null);
        }

        $this->setCache($index, $value);

        $request = $this->request(
            "
            INSERT INTO `{$this->table}` 
            SET `key` = :index, `value` = :value 
            ON DUPLICATE KEY UPDATE `value` = :value
        ",
            [
                'index' => $index,
                '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;
    }

$ this->リクエストは、データを非同期に書き込むPromiseを作成します。また、キャッシュを使用した操作は同期して行われます。つまり、データベースへの書き込みを待つことはできません。同時に、読み取り操作がすぐに新しいデータを返し始めることを確認してください。



amphpのonResolveメソッドは非常に便利であることがわかりました。挿入が完了すると、データは再びキャッシュに書き込まれます。一部の書き込み操作が遅れ、キャッシュとベースが異なり始めた場合、キャッシュは最後にベースに書き込まれた値で更新されます。それら。キャッシュは再びベースと一致するようになります。

ソース

→プルリクエストへのリンク



そして、そのように別のユーザーがpostgreのサポートを追加しました。そのための指示を書くのにたった5分しかかかりませんでした。



重複するメソッドを一般的な抽象クラスSqlArrayに移動することで、コードの量を減らすことができます。

もう一つ

テレグラムからメディアファイルをダウンロードしている間、標準のガベージコレクターphpは作業に対応せず、ファイルの一部がメモリに残っていることに気づきました。通常、リークはファイルと同じサイズでした。考えられる原因：10,000個のリンクが蓄積されると、ガベージコレクターが自動的にトリガーされます。私たちの場合、リンクは数（数十）でしたが、それぞれがメモリ内のメガバイトのデータを参照する可能性があります。 mtprotoの実装で数千行のコードを研究するのは非常に怠惰でした。最初に\ gc_collect_cycles（）;でエレガントなクラッチを試してみませんか？



驚いたことに、それは問題を解決しました。これは、定期的なクリーニングの開始を構成するだけで十分であることを意味します。幸い、amphpは、指定された間隔でバックグラウンドで実行するためのシンプルなツールを提供します。



毎秒メモリをクリアするのは簡単すぎてあまり効果的ではないように見えました。前回のクリーンアップ以降のメモリゲインをチェックするアルゴリズムに落ち着きました。ゲインがしきい値よりも大きい場合、クリアが発生します。

<?php

namespace danog\MadelineProto\MTProtoTools;

use Amp\Loop;
use danog\MadelineProto\Logger;

class GarbageCollector
{
    /**
     * Ensure only one instance of GarbageCollector
     * 		when multiple instances of MadelineProto running.
     * @var bool
     */
    public static bool $lock = false;

    /**
     * How often will check memory.
     * @var int
     */
    public static int $checkIntervalMs = 1000;

    /**
     * Next cleanup will be triggered when memory consumption will increase by this amount.
     * @var int
     */
    public static int $memoryDiffMb = 1;

    /**
     * Memory consumption after last cleanup.
     * @var int
     */
    private static int $memoryConsumption = 0;

    public static function start(): void
    {
        if (static::$lock) {
            return;
        }
        static::$lock = true;

        Loop::repeat(static::$checkIntervalMs, static function () {
            $currentMemory = static::getMemoryConsumption();
            if ($currentMemory > static::$memoryConsumption + static::$memoryDiffMb) {
                \gc_collect_cycles();
                static::$memoryConsumption = static::getMemoryConsumption();
                $cleanedMemory = $currentMemory - static::$memoryConsumption;
                Logger::log("gc_collect_cycles done. Cleaned memory: $cleanedMemory Mb", Logger::VERBOSE);
            }
        });
    }

    private static function getMemoryConsumption(): int
    {
        $memory = \round(\memory_get_usage()/1024/1024, 1);
        Logger::log("Memory consumption: $memory Mb", Logger::ULTRA_VERBOSE);
        return (int) $memory;
    }
}