2007-06-04

1. コネクションプール

現在進行中。ここの実装の良し悪しは性能に多大なインパクトを与えるので、重要。
昨日のエントリで書いたepoll/kqueueクラスを核に実装。

class ASStreamManager {    // pimplイディオムで
public:
    void registHandler(uint8_t magic, ASHandlerFunction* new_handler);
    ASSocket negotiate(const NodeAddress& to, ASSendFunction* proc);
public:
    std::list<DynamicConfigurator> getConfigurator(void);
};

インターフェイスは単純で、registHandler()で関数オブジェクトを登録しておくと、先頭の1バイトがmagicなパケットを受け取ったときに、new_handlerを呼んでもらえる。こちらから他のノードにデータを送りたいときは、negotiate()を呼ぶと、toと接続されたソケットを返してもらえる。
インターフェイスは単純でも中身は大変。

DynamicConfiguratorというのは、実行中に設定を変更するためクラス。UNIXドメインソケットなどから専用クライアントプログラムを通じてユーザーからのコマンドを受け付けて、設定を動的に変更する。

class DynamicConfigurator {    // pimplで
public:
    const std::string& getName(void);
public:
    ConfigurationViewFormat set(const std::string& key, std::vector<std::string> argv);
    ConfigurationViewFormat get(const std::string& name);
};

専用クライアント（対人間用）の動作イメージ:

Command> get TCP::MaxConnections
100
Command> set TCP::MaxConnections 1000
    # DynamicConfigurator::getName() == "TCP" のオブジェクトの、set("MaxConnections", {"1024"});を呼ぶ
TCP::MaxConnections is now 1000.    # ココの部分の表示/ユーザーからの入力はConfigurationViewFormatクラスで制御
Restarting ASStreamManager.
Command> help  # 組み込みコマンド help
Modules:
TCP      # 1つのDynamicConfiguratorオブジェクトが1つのModuleと対応
Storage  #あるクラスが複数のDynamicConfiguratorをサポートしてもいい
Files
Nodes

Type `help <module_name>' for more information.
Command>

こういう仕組みは最初から頭に入れておいて、動的な設定変更が可能なように実装していかないと、後から追加するのは難しい。

2. ストレージインターフェイスを作る

２つのIOインターフェースの内のストレージ側。（もう片方はアクセスインターフェイス）

template <typename IMPL>    // 継承は使わないで、ひたすらインライン化でパフォーマンスを稼ぐ
class StorageInterface {
public:
    StorageInterface(IMPL& impl);
public:
    void direct_write(key_type key, const char* buffer, uint64_t offset, uint64_t len);
    // void atomic_write(key_type key, const char* buffer, uint64_t offset, uint64_t len);
public:
    void direct_read(key_type key, char* buffer, uint64_t offset, uint64_t len);
    // void atomic_read(key_type key, char* buffer, uint64_t offset, uint64_t len);
    // void filtered_read(...);    // 後回し
private:
    IMPL& impl;
// DynamicConfiguration不可  （実行途中でIMPLを入れ替えたりはできない）
//     → ストレージインターフェイスは全ノードで同じものを使う
};

単なるインターフェイスで、IMPLのメソッドを呼び出すだけ。IMPLはコンパイル時に切り替えられる。分散ファイルシステム用ファイルシステムをストレージに使う実装を作る。
テスト用にはコンソールをストレージにする？writeは標準出力に出力。readは、（"offset = %s, len = %s", offset, len）とパディングを埋めたもの返す。こういうのはストレージとは言わないか。テストには最適。

3. ノード一覧キャッシュクラスを作る

どのノードがどこのデータを持っているかというテーブルのキャッシュ。キャッシュなので不正確でも良い。

class NodeCache {    // pimplで
public:
    NodeAddress searchOneNode(key_type key);
    template <typename Func> void searchAllNodes(key_type key, Func callback);   // 実装1
    void searchAllNodes(key_type key, std::vector<NodeAddress>& result_nodes);   // 実装2
    void searchAllNodes(key_type key, MultiNodeAddress& result_nodes);    // 実装3
private:
    void addNode(const NodeAddress& node);
    void addKey(const NodeAddress& node, key_type key);
    void removeNode(const NodeAddress& node);
    void removeKey(const NodeAddress& node, key_type key);
// privateでASHandlerがいろいろ
};

searchAllNodesの実装1は、NodeAddressのコピーは必要無いけど、ロック時間が長い。実装2は、コピーが必要だけど、ロック時間が短い。うーむ。

複数のNodeAddressを管理するのに効率がいいNodeAddressクラスを作ればいいかも。今のNodeAddressの構造は↓で、データを全部一つのバイト列で持っているという超効率重視の設計。

class NodeAddress {
public:
    inline isIPv6(void) const;
    inline const struct in6_addr* getin6addr(void) const throw();
    inline const struct in_addr*  getin4addr(void) const throw();
    inline uint16_t getPortByNetworkByteOrder(void) const throw();
    inline const char* getRaw(void) const throw();
private:
    char m_data[19];
};

ということは、複数のNodeAddressを管理するには、単純に長いバイト列を用意すればいいわけで。

class MultiNodeAddress {
public:
    MultiNodeAddress(MemPool& pool) : m_data( pool.malloc_scoped_strea() ) {}
public:
    inline void addNode(const NodeAddress& node);
public:
    inline const NodeAddress& operator[] (size_t n);
private:
    MemPool::scoped_stream m_data;
    size_t m_num;
public:
    // operator=をmemcpyで自作
};

NodeCacheは、MultiNodeAddressでノード群を保持する。
今メモリプール付きスレッドプール（スレッドプールのそれぞれのスレッドごとに、メモリプールが付いている）を作っていて、これを全面的に採用するので、どこからでもシングルスレッドのメモリプール（とても速い）が使える。メモリプールのチャンクサイズはSTREAM_BUFFER_SIZEで、たぶん16KB。これなら複数のNodeAddressを返したいときでも、コピーはmemcpy一発で、そこそこ速いはず。でもこれを実装する手間に見合うだけの効果があるのかどうか…。

4. readを処理するクラスを作る

アクセスインターフェースから要求を受け取り、NodeCacheを参照してどこのノードがデータを持っているか探して、ダウンロードしてくる。
他のノードからread要求が来たら、それに応える。

class ReadProcessor {
public:
    void direct_read(key_type key, char* buffer, uint64_t offset, uint64_t len);
private:
    class reqRead : public ASStreamManager::ASHandlerFunction {
        void operator() (ASSocket& socket);
    };
};

5. ネットワーク全体のハッシュ空間の管理クラスを作る

NodeAddress→NodeIDの対応表と、NodeIDの一覧を持つ。あるkeyを探したいときに、そのkeyを誰が持っているかを知っているノードは、この対応表を参照すると分かる。

class HashSpace {
public:
    NodeID searchOneNode(key_type key);
    void searchAllNodes(key_type, std::valarray<NodeID>& result_nodes);
        // あるkeyを誰が持っているかの情報はプライマリ-セカンダリ型で冗長化されているので、複数のノードが知っている
    NodeAddress getNodeAddress(NodeID id);
private:
    void addNode(NodeAddress& node, NodeID id, key_type key);
// ASHandlerがいろいろ
};

6. 自分の管理しているkeyの管理クラス

「あるkeyを誰が持っているかの情報」を管理するクラス。

class KeysPrimary {
public:
    void nodes(key_type key, std::vector<NodeAddress>& result_nodes);
// ASHandlerがいろいろ
};

7. keyの管理クラスのセカンダリ

「あるkeyを誰が持っているかの情報」は、プライマリ-セカンダリ型で冗長化する。プライマリは↑で、セカンダリが↓。１台のノードが複数のプライマリに対するセカンダリだったりするので、このクラスは複数のインスタンスを持つ。

class KeysSecondary {
// アクセサ無し
// ASHandlerがいろいろ
};

困った点が１つ。key_typeを何にするか。つまり、分散連想配列はkeyとvalueの組だけど、valueはバイト列（char* buf + uint64_t len）として、keyの型は何にするか。ファイルシステムなどなどを作るときにはkeyがPATHになると思うので、keyの型はバイト列でないと困るけど、それだとメモリを食って大変。トラフィックも増えるし。uint64_tにしてしまいたい…。うーむ。

Blog by Sadayuki Furuhashi

Treasure Data MessagePack Embulk Fluentd @twitter

分散ファイルシステムクラス実装計画メモ