xmisao -- ぺけみさお

RubyでAmazon BedrockのClaude 3 Opusを使う

Sat, 27 Apr 2024 00:00:00 -0500

はじめに

先日Claude 3 OpusがAmazon Bedrockで使用可能になった。

https://aws.amazon.com/jp/blogs/news/anthropics-claude-3-opus-model-on-amazon-bedrock/

JavaScriptやPythonの例ばかりでRubyで完全に動作するコードが見あたらなかったので、RubyのAWS SDKからのOpusの呼び出し方法を書いておく。

前提ソフトウェア

ソフトウェア	バージョン	備考
Ruby	3.3.0	-
aws-sdk-bedrockruntime	1.7.0	-

下準備

Bedrockでは各モデルの利用前にAWSマネジメントコンソールからアクセスをリクエストして承認される必要がある。その辺はクラメソの記事などを参照。自分でもやってみたところ、すぐ承認された。

https://dev.classmethod.jp/articles/claude-3-opus-bedrock/

Opusを呼び出す

Gemfile

source "https://rubygems.org"

gem "aws-sdk-bedrockruntime"

gem "rexml"
# gem "nokogiri"

# gem "base64"

ポイントは以下のとおり。

aws-sdk-bedrockruntime がモデルを実行するためのGem。
- aws-sdk-bedrock もあるが、こちらはBedrockの構成を管理するAPIのためのGemで別物なことに注意。
rexml や nokogiri などいずれかのXMLライブラリが必要。
- 入れないとAWS SDKの呼び出し時に次のエラーになる。 Unable to find a compatible xml library. Ensure that you have installed or added to your Gemfile one of ox, oga, libxml, nokogiri or rexml (RuntimeError)
Ruby3.3だと警告が表示されるので、標準添付ライブラリの base64 もGemの方を入れておくのが良い。
- warning: base64 was loaded from the standard library, but will no longer be part of the default gems since Ruby 3.4.0. Add base64 to your Gemfile or gemspec. Also contact author of aws-sdk-core-3.193.0 to add base64 into its gemspec.

プロンプトに対して回答を得るスクリプト

require "aws-sdk-bedrockruntime"

bedrock_client = Aws::BedrockRuntime::Client.new(
  region: 'us-west-2', # Opusが提供されているオレゴンリージョンを指定
  http_read_timeout: 600.0 # 応答に時間がかかるとタイムアウトするので600秒に延長しておく
)

prompt = "こんにちは。"

payload = {
  messages: [
    {
      role: "user",
      content: [
        {
          type: "text",
          text: prompt
        }
      ]
    }
  ],
  max_tokens: 1000,
  anthropic_version: "bedrock-2023-05-31"
}

body = payload.to_json

resp = bedrock_client.invoke_model({
  model_id: "anthropic.claude-3-opus-20240229-v1:0",
  content_type: "application/json",
  body:
})

resp #=> #<struct Aws::BedrockRuntime::Types::InvokeModelResponse body=#<StringIO:0x00007ffae6d78ed8>, content_type="application/json">
resp.body #=> #<StringIO:0x00007ffae6d78ed8>

resp_body = resp.body.read

resp_json = JSON.parse(resp_body)
# {"id"=>"msg_bdrk_01JdUa8GaBKC9T9FLnM5mbtj",
#  "type"=>"message",
#  "role"=>"assistant",
#  "content"=>[{"type"=>"text", "text"=>"こんにちは。今日はどんなことでお話ししましょうか?私にできることがあれば、何でも聞いてくださいね。"}],
#  "model"=>"claude-3-opus-20240229",
#  "stop_reason"=>"end_turn",
#  "stop_sequence"=>nil,
#  "usage"=>{"input_tokens"=>13, "output_tokens"=>47}}

puts resp_json["content"][0]["text"]
#=> こんにちは。今日はどんなことでお話ししましょうか?私にできることがあれば、何でも聞いてくださいね。

bodyパラメータとして送っているJSONの形式がClaude 2以前と異っていて、その情報が少なくてハマった。 AnthropicのMessages APIという形式でないといけない。 JSONの形式については以下のAWSユーザーガイドを見ると良い模様。

https://docs.aws.amazon.com/bedrock/latest/userguide/model-parameters-anthropic-claude-messages.html

古い形式のbodyパラメータでClaude 3を呼ぼうとすると "claude-3-opus-20240229" is not supported on this API. Please use the Messages API instead. (Aws::BedrockRuntime::Errors::ValidationException) といった例外が発生する。

invoke_model で model_id として指定できる値についてもAWSユーザーガイドの一覧を参照。別の値を指定することで他のモデルを呼ぶことができる。現時点でHaikuやSonnetを含むClaude 3のオンデマンドスループットのIDを抜粋すると以下のようになっていた。

https://docs.aws.amazon.com/bedrock/latest/userguide/model-ids.html

Anthropic	Claude 3 Sonnet	1.0	anthropic.claude-3-sonnet-20240229-v1:0
Anthropic	Claude 3 Haiku	1.0	anthropic.claude-3-haiku-20240307-v1:0
Anthropic	Claude 3 Opus	1.0	anthropic.claude-3-opus-20240229-v1:0

なおこの例ではシンプルに invoke_model を使っているが、コールバックで応答が得られる invoke_model_with_response_stream もある。要件によってはこちらを使った方が良いだろう。

invoke_model はOpusの場合、応答が返るまで長時間かかることがあるため http_read_timeout を長く設定しておくと良い。デフォルトでは60秒応答がないと Net::ReadTimeout となってしまう。

BedrockRuntime::Client の使い方の詳細はAWS SDKのドキュメントを参照。

https://docs.aws.amazon.com/sdk-for-ruby/v3/api/Aws/BedrockRuntime/Client.html

実行する

特別なことはない。 AWSのクレデンシャルが必要で、典型的な実行方法は以下のとおり。

モデルの呼び出しには bedrock:InvokeModel アクションが許可されている必要がある。

# デフォルトのAWSプロファイルやIAMロールの場合は何も指定せず実行
bundle exec ruby sample.rb
 
# 特定のAWSプロファイルxxxを使う場合はAWS_PROFILE環境変数を指定
AWS_PROFILE=xxx bundle exec ruby sample.rb

# アクセスキーを使う場合はAWS_ACCESS_KEY_IDとAWS_SECRET_ACCESS_KEY環境変数を指定
AWS_ACCESS_KEY_ID=xxx AWS_SECRET_ACCESS_KEY=xxx bundle exec ruby sample.rb

まとめ

このように、ごく僅かなコーディングで最新の基盤モデルであるOpusをRubyから呼び出すことができた。

日本語が扱える実用的な基盤モデルが追加されたことで、AWS上で動作している既存Railsアプリケーションに生成AIを使った機能を組み込む、といったことも従来より気軽にできるようになるだろう。

生成AIは群雄割拠でAWSにどこまで拘るかは悩ましい状況だが、Bedrockは他のAWSサービスと同じように構成できIAMで権限を管理できる等、インフラや運用の面でもメリットがあるように思える。

LinuxとWindowsでビルドできるRust-SDL2プロジェクトを作成する

Mon, 11 Jan 2021 00:00:00 -0600

概要

Rust-SDL2ではSDL2を使ったアプリケーションを複数のプラットフォーム向けにビルドできます。 cargo-vcpkgも併用すればvcpkgでSDL2をインストールできるため、事前にライブラリをインストールする必要もありません。ビルドできるバイナリでSDL2は静的リンクになります。

この記事ではRust-SDL2とcargo-vcpkgを使った最小限のプロジェクトを作成し、LinuxとWindowsの両方でビルドし動作させることができるか試してみます。ソースコード一式は以下のGitHubリポジトリにあります。

https://github.com/xmisao/sdl2example

前提ソフトウェア

ソフトウェア	バージョン	備考
Rust	1.49.0	-
Rust-SDL2	0.34.2	https://github.com/Rust-SDL2/rust-sdl2
cargo-vcpkg	0.15.0	https://github.com/mcgoo/vcpkg-rs

準備

Rust, Gitのインストール

RustとGitをインストールします。 Gitはcargo-vcpkgが動作するために必要です。公式も親切ですし、ググれば出てくるので、具体的な手順は割愛します。

cargo-vcpkgのインストール

以下のコマンドを実行しcargo-vcpkgをインストールします。

cargo install cargo-vcpkg

これでcargo vcpkgというサブコマンドが使えるようになります。

アプリケーションの実装

cargoパッケージを作成する

バイナリテンプレートで適当なcargoパッケージを作成します。

cargo new --bin sdl2example

以降の作業はcargoパッケージのルートディレクトリで行います。

依存関係の設定

Rust-SDL2でSDL2をcargo-vcpkgでインストールするように依存関係を設定します。設定はRust-SDL2のREADMEを参考にします。

https://github.com/Rust-SDL2/rust-sdl2#windows-linux-and-macos-with-vcpkg

以下を Cargo.toml の末尾に追記します。

[dependencies.sdl2]
version = "0.34"
default-features = false
features = ["ttf","image","gfx","mixer","static-link","use-vcpkg"]

[package.metadata.vcpkg]
dependencies = ["sdl2", "sdl2-image[libjpeg-turbo,tiff,libwebp]", "sdl2-ttf", "sdl2-gfx", "sdl2-mixer"]
git = "https://github.com/microsoft/vcpkg"
rev = "a0518036077baa4"

[package.metadata.vcpkg.target]
x86_64-pc-windows-msvc = { triplet = "x64-windows-static-md" }

main.rsを実装する

以下のソースコードはRust-SDL2のデモでウィンドウを出すだけのSDL2アプリケーションです。 main.rsを丸々この内容に書き換えます。

extern crate sdl2;

use sdl2::event::Event;
use sdl2::keyboard::Keycode;
use sdl2::pixels::Color;
use std::time::Duration;

pub fn main() -> Result<(), String> {
    let sdl_context = sdl2::init()?;
    let video_subsystem = sdl_context.video()?;

    let window = video_subsystem
        .window("rust-sdl2 demo: Video", 800, 600)
        .position_centered()
        .opengl()
        .build()
        .map_err(|e| e.to_string())?;

    let mut canvas = window.into_canvas().build().map_err(|e| e.to_string())?;

    canvas.set_draw_color(Color::RGB(255, 0, 0));
    canvas.clear();
    canvas.present();
    let mut event_pump = sdl_context.event_pump()?;

    'running: loop {
        for event in event_pump.poll_iter() {
            match event {
                Event::Quit { .. }
                | Event::KeyDown {
                    keycode: Some(Keycode::Escape),
                    ..
                } => break 'running,
                _ => {}
            }
        }

        canvas.clear();
        canvas.present();
        ::std::thread::sleep(Duration::new(0, 1_000_000_000u32 / 30));
        // The rest of the game loop goes here...
    }

    Ok(())
}

ビルドと実行

ビルド

アプリケーションをビルドします。

最初はcargo vcpkg buildでSDL2を落としてきてビルドしてやる必要があります。

cargo vcpkg build

あとは通常通りビルドするだけです。リリースビルドしています。

cargo build --release

これはtarget/release以下に実行ファイルが出力されます。容量はWindowsで3.4MB, Linuxで4.6MBほどでした。

補足: ダイナミックリンクされたライブラリの確認

この方法ではSDL2は静的リンクされます。 Linuxでlddで確認してみます。

$ ldd target/release/sdl2example
        linux-vdso.so.1 (0x00007fff7777f000)
        libgcc_s.so.1 => /lib/x86_64-linux-gnu/libgcc_s.so.1 (0x00007f66b485d000)
        libpthread.so.0 => /lib/x86_64-linux-gnu/libpthread.so.0 (0x00007f66b483b000)
        libm.so.6 => /lib/x86_64-linux-gnu/libm.so.6 (0x00007f66b46f7000)
        libdl.so.2 => /lib/x86_64-linux-gnu/libdl.so.2 (0x00007f66b46f1000)
        libc.so.6 => /lib/x86_64-linux-gnu/libc.so.6 (0x00007f66b452c000)
        /lib64/ld-linux-x86-64.so.2 (0x00007f66b4a4d000)

実行

ビルドしたアプリケーションを実行します。真っ赤なウィンドウが出たら成功です。

cargo run --release

Windowsの動作例

Linuxの動作例

まとめ

かなり簡単にSDL2を使ったアプリケーションを作ることができました。

Rustとcargo-vcpkgを使った開発体験もとても良いと思います。SDL2でゲームが作りたければ活用してみても良いのではないでしょうか。

ただしRust-SDL2にはまだ参考にできるドキュメントが少ないです。狙い通り動かすにはCでの使い方を知った上で適宜Rust-SDL2のコードを読む必要があるでしょう。

Redisのトランザクションと楽観ロックについて調べたのでRubyから使ってみた

Sun, 12 Apr 2020 00:00:00 -0500

概要

たまたま使う機会があって調べたのでRedisのトランザクションをRubyのredisGemから扱う方法をまとめます。通常RubyからRedisを使うケースはSidekiqのようなGemやRailsのRedisキャッシュストア, ActionCable経由で使うことがほとんで、直接Redisを意識して使うケースは希かも知れません。

TL;DR

RedisコマンドのMULTI, EXECで複数のコマンドをアトミックに実行できます。
アトミックとはいえロールバックされません。コーディングミスによる誤った操作で部分的にコマンドが実行されることはあり得ます。
さらにRedisコマンドのWATCHを使うと楽観ロックによる排他制御が行えます。

Redisのトランザクションについての詳細についてはRedisのドキュメントを参照されることをおすすめします。

Transactions – Redis

前提ソフトウェア

ソフトウェア	バージョン	備考
Redis	5.0.8
ruby	2.6.3p62
redis	4.1.3	https://rubygems.org/gems/redis/versions/4.1.3

複数のコマンドをアトミックに実行する

例としてキーに値を設定(SET)してTTLを設定(EXPIRE)する操作を考えます。このようなケースではSETした直後にスクリプトが異常終了してEXPIREが実行できない場合SETだけ実行されるとTTLが設定されないキーができてしまい不整合が生じます。

複数のコマンドをアトミックに実行したい場合はRedisのMULTI, EXECコマンドを使います。RedisにおいてMULTI以降のコマンドはキューイングされEXECした時にアトミックに実行されます。Redisは直列で処理を行うので他のクライアントの操作に対して割り込まれることもありません。

MULTI
SET mykey, 'abc'
TTL mykey, 60
EXEC

MULTIはEXECせずにDISCARDすると中止することができます。中止するとそれまでキューイングされたコマンドは実行されません。

MULTI
SET mykey, 'abc'
TTL mykey, 60
DISCARD

これでSET, TTLのような一連の書き込み処理をまとめて安全に実行できます。なおコマンド実行はキューイングされますのでGETのような読み込みコマンドはMULTIで実行できません。読み込みを伴う排他制御は後述のWATCHによる楽観ロックを使います。

上記のコマンドを実行するスクリプトをRubyのredisGemを使って実装してみます。Redis#multiは呼び出し時のブロックの有無により動作が異なります。

参考: https://www.rubydoc.info/github/redis/redis-rb/Redis:multi

Redis#multiをブロックなしで呼び出すとMULTIが単独で実行されます。EXECやDISCARDはRedis#execやRedis#discardを使って実行してやる必要があります。

require 'redis'
redis = Redis.new

p redis.multi #=> "OK"
p redis.set('mykey', 'abc') #=> "QUEUED"
p redis.expire('mykey', 60) #=> "QUEUED"
p redis.exec #=> ["OK", 1]

Redis#multiをブロック付きで呼び出すとブロックの内部はEXEC後に実行されブロックを抜ける際に自動的にEXECされます。Redis#multiはredis gemのパイプライン機能で一括に実行されます。Redis#multiブロック中の操作はRedis::Futureが返りRedisへの操作はブロックを抜けるまで保留されます。もしブロック内の処理でRubyの例外が発生した場合はそもそもMULTIさえ実行されずRedisの操作は一切行われません。

require 'redis'

redis = Redis.new

res = redis.multi do
  p redis.set('mykey', 'abc') #=> <Redis::Future [:set, "mykey", "abc"]>
  p redis.expire('mykey', 60) #=> <Redis::Future [:expire, "mykey", 60]>
end
p res #=> ["OK", true]

注意点

RedisのEXECはアトミックですが万一EXECに失敗してもロールバックはされません。このためプログラムミスで不適切なコマンドを実行しようとして失敗した場合、MULTI以降にキューイングしたコマンドが部分的に実行され得ることに注意が必要です。

以下はString型の値を持つキーに対してList型にしか実行できないLPOPを実行する例です。文法的には正しいですがEXECは失敗して例外Redis::CommandErrorが発生します。LPOPの前後に実行しているSETは実行済みの状態のままになります。

require 'redis'

redis = Redis.new

begin
  redis.multi do
    redis.set('a', 'partial execution')
    redis.lpop('a')
    redis.set('b', 'partial execution')
  end
rescue
  p $! #=> #<Redis::CommandError: WRONGTYPE Operation against a key holding the wrong kind of value>
end

p redis.get('a') #=> "partial execution"
p redis.get('b') #=> "partial execution"

もう1つのケースとしてRedisサーバがEXECの最中にクラッシュしたらどうなるのでしょうか。これは私も経験したことがないのですが、もしRedisサーバがクラッシュした場合でappend-only fileを使っている場合(appendonly yes)はエラーを検出して回復できるそうです。

詳細はRedisのドキュメントを参照して下さい。なぜRedisにはロールバックがないのかを含めて解説されています。

楽観ロックによる排他制御を行う

Redisのドキュメントに倣って、特定のキーの値を2,000回ほどGETとSETでスクリプトからカウントアップすることを考えます。(RedisにはINCRコマンドがありますので本当にキーの値をカウントアップする時はそちらを使った方が良いです)

楽観ロックを使わず素朴に書いた以下のスクリプトを実行してみます。

require 'redis'

Redis.new.set('counter', 0)

threads = 2.times.map do
  Thread.start do
    redis = Redis.new

    1000.times do
      val = redis.get('counter').to_i + 1
      redis.set('counter', val)
    end
  end
end

threads.each(&:join)

p Redis.new.get('counter') #=> "1306"

当然ながら排他制御されていませんので2,000より少ない数字が出力されます。値を読んだ後に別スレッドで値が書き込まれ、結果的に同じ値を書いてしまうことがあるからです。

このような場合はRedisでMULTI, EXEC に加えて WATCH コマンドを使うといわゆる楽観ロックによる排他制御が行えます。WATCHを使うとWATCHしたキーがEXECまでに間に他のクライアントによって変更されるとEXECは失敗します。楽観ロックの対象はWATCHで指定したキーで、その後に実際にキーを読むかは関係ありません。

WATCHはRedis#watchで実行できます。Redis#watchもRedis#multiと同じようにブロック有無によって挙動が異なります。Redis#watchをブロック付きで呼んだ場合は、Redis#watchはブロックの評価結果を戻り値として返します。またRedis::ConnectionErrorを除くStandardErrorのサブクラスの例外がブロック内で発生すると自動的にUNWATCHを実行してWATCHを解除してくれます。

参考: https://www.rubydoc.info/github/redis/redis-rb/Redis:watch

元のRubyスクリプトを改良しWATCH, MULTI, EXECする形に書き直してみます。以下のスクリプトではRedis#multiで実行するEXECが失敗した場合にRedis#watchの戻り値としてnilが得られます。これをチェックしてredoで無限リトライするようにします。実際にはリトライ回数を制限するのが良いでしょう。

require 'redis'

Redis.new.set('counter', 0)

threads = 2.times.map do
  Thread.start do
    redis = Redis.new

    1000.times do
      res = redis.watch('counter') do
        val = redis.get('counter').to_i + 1

        redis.multi do
          redis.set('counter', val)
        end
      end

      redo unless res
    end
  end
end

threads.each(&:join)

p Redis.new.get('counter') #=> "2000"

期待どおり2,000までカウントアップすることができました。

Tips: RubyクライアントからRedisに実行されているコマンドを確認する

RubyのクライアントからRedisに対してどのようなコマンドが実行されているのか確認したくなることがあります。特にRedisインスタンスのブロック付きメソッドを使用しているとパイプライン機能で実行されるコマンドをRubyスクリプト中から確認することは困難です。このようなときはRedisのMONITORコマンドを使用するとRedisサーバで実行したコマンドを確認してデバッグすることができます。

以下はredis-cliからMONITORコマンドを実行して適当な操作を行った際に得られる出力の例です。

# redis-cli
127.0.0.1:6379> MONITOR
OK
1586676376.691487 [0 172.28.0.1:38078] "multi"
1586676376.692826 [0 172.28.0.1:38078] "set" "foo" "bar"
1586676376.692937 [0 172.28.0.1:38078] "incr" "baz"
1586676376.692956 [0 172.28.0.1:38078] "exec" 

参考: https://redis.io/commands/monitor

外部キー制約をかける時にMySQLはインデックスを暗黙的に作成するがPostgreSQLは作成しない

Sun, 29 Mar 2020 00:00:00 -0500

概要

外部キー制約をかける時にMySQLはインデックスを暗黙的に作成しますがPostgreSQLは作成しません。この違いを知らないと、同じスキーマを作成したつもりであってもインデックス有無が異なるため、MySQLとPostgreSQLでパフォーマンスに大きな違いが生じるかも知れません。

前提ソフトウェア

ソフトウェア	バージョン	備考
MySQL	8.0.19	-
PostgreSQL	12.2	-

ドキュメントを確認する

ドキュメントにもそのように書いてあります。現時点で一番新しいバージョンの日本語と英語のドキュメントを引用します。(ドキュメントが変更されていて、日本語と英語は一部対応していないところがあります) 余談ですがPostgreSQLのユーザに選択を委ねる姿勢の方が私は好きです。

https://dev.mysql.com/doc/refman/5.6/ja/create-table-foreign-keys.html

index_name は、外部キー ID を表します。外部キーをサポートできる子テーブル上に明示的に定義されたインデックスがすでに存在する場合、 index_name 値は無視されます。それ以外の場合、MySQL は、次のルールに従って名前が付けられた外部キーのインデックスを暗黙的に作成します。

定義されている場合は、CONSTRAINT symbol 値が使用されます。それ以外の場合は、FOREIGN KEY index_name 値が使用されます。

CONSTRAINT symbol と FOREIGN KEY index_name のどちらも定義されていない場合、外部キーのインデックス名は、参照している外部キーカラムの名前を使用して生成されます。

https://dev.mysql.com/doc/refman/8.0/en/create-table-foreign-keys.html

MySQL requires indexes on foreign keys and referenced keys so that foreign key checks can be fast and not require a table scan. In the referencing table, there must be an index where the foreign key columns are listed as the first columns in the same order. Such an index is created on the referencing table automatically if it does not exist. This index might be silently dropped later if you create another index that can be used to enforce the foreign key constraint. index_name, if given, is used as described previously.

https://www.postgresql.jp/document/11/html/ddl-constraints.html#DDL-CONSTRAINTS-FK

外部キーは主キーであるかまたは一意性制約を構成する列を参照しなければなりません。これは、被参照列は常に(主キーまたは一意性制約の基礎となる)インデックスを持つことを意味します。このため、参照行に一致する行があるかどうかのチェックは効率的です。被参照テーブルからの行のDELETEや被参照行のUPDATEは、古い値と一致する行に対して参照テーブルのスキャンが必要となるので、参照行にもインデックスを付けるのは大抵は良い考えです。これは常に必要という訳ではなく、また、インデックスの方法には多くの選択肢がありますので、外部キー制約の宣言では参照列のインデックスが自動的に作られるということはありません。

https://www.postgresql.org/docs/12/ddl-constraints.html#DDL-CONSTRAINTS-FK

A foreign key must reference columns that either are a primary key or form a unique constraint. This means that the referenced columns always have an index (the one underlying the primary key or unique constraint); so checks on whether a referencing row has a match will be efficient. Since a DELETE of a row from the referenced table or an UPDATE of a referenced column will require a scan of the referencing table for rows matching the old value, it is often a good idea to index the referencing columns too. Because this is not always needed, and there are many choices available on how to index, declaration of a foreign key constraint does not automatically create an index on the referencing columns.

実験

では実際にMySQL, PostgreSQLの両方にスキーマを作成して試してみます。スキーマは共通で以下のSQLを使います。インデックスが作成されるかを確認するためのスキーマなので最小限のものです。

CREATE TABLE parents (
  id INT NOT NULL,
  PRIMARY KEY(id)
);

CREATE TABLE children (
  id INT NOT NULL,
  parent_id INT NOT NULL,
  PRIMARY KEY (id),
  FOREIGN KEY (parent_id) REFERENCES parents (id)
);

MySQL

MySQLでchildrenテーブルのインデックスを確認してみます。parent_idに対して暗黙的にインデックスが作成されていることがわかります。

mysql> SHOW INDEX FROM children;
+----------+------------+-----------+--------------+-------------+-----------+-------------+----------+--------+------+------------+---------+---------------+---------+------------+
| Table    | Non_unique | Key_name  | Seq_in_index | Column_name | Collation | Cardinality | Sub_part | Packed | Null | Index_type | Comment |　Index_comment | Visible | Expression |
+----------+------------+-----------+--------------+-------------+-----------+-------------+----------+--------+------+------------+---------+---------------+---------+------------+
| children |          0 | PRIMARY   |            1 | id          | A         |           0 |     NULL |   NULL |      | BTREE      |         |               | YES     | NULL       |
| children |          1 | parent_id |            1 | parent_id   | A         |           0 |     NULL |   NULL |      | BTREE      |         |               | YES     | NULL       |
+----------+------------+-----------+--------------+-------------+-----------+-------------+----------+--------+------+------------+---------+---------------+---------+------------+
2 rows in set (0.00 sec)

テーブルが空なのでなんともですが、ついでにparent_idでchildrenを絞り込むクエリの実行計画を見てみます。するとparent_idをキー(key)とする実行計画となっていて、暗黙的に作成されたインデックスが使われてる高速なクエリとなることがわかります。

mysql> EXPLAIN SELECT * FROM children WHERE parent_id = 999;
+----+-------------+----------+------------+------+---------------+-----------+---------+-------+------+----------+-------------+
| id | select_type | table    | partitions | type | possible_keys | key       | key_len | ref   | rows | filtered | Extra       |
+----+-------------+----------+------------+------+---------------+-----------+---------+-------+------+----------+-------------+
|  1 | SIMPLE      | children | NULL       | ref  | parent_id     | parent_id | 4       | const |    1 |   100.00 | Using index |
+----+-------------+----------+------------+------+---------------+-----------+---------+-------+------+----------+-------------+
1 row in set, 1 warning (0.00 sec)

PostrgreSQL

同様にPostgreSQLに対するインデックスを確認してみます。Foreign-key constrainsはかかっていますが、Indexesはidに対するchildren_pkeyのみとなっています。

postgres=# \d children;
               Table "public.children"
  Column   |  Type   | Collation | Nullable | Default
-----------+---------+-----------+----------+---------
 id        | integer |           | not null |
 parent_id | integer |           | not null |
Indexes:
    "children_pkey" PRIMARY KEY, btree (id)
Foreign-key constraints:
    "children_parent_id_fkey" FOREIGN KEY (parent_id) REFERENCES parents(id)

こちらもparent_idでchildrenを絞り込むクエリの実行計画を見てみます。Seq Sacnですのでシーケンシャルスキャンになっています。

postgres=# EXPLAIN SELECT * FROM children WHERE parent_id = 999;
                        QUERY PLAN
----------------------------------------------------------
 Seq Scan on children  (cost=0.00..38.25 rows=11 width=8)
   Filter: (parent_id = 999)
(2 rows)

念のためparent_idにインデックスを追加して試してみます。

CREATE INDEX ON children (parent_id);

実行計画もBitmap Heap Scanになりました。children_parent_id_idxが使われていることがわかります。

postgres=# EXPLAIN SELECT * FROM children WHERE parent_id = 999;
                                      QUERY PLAN
--------------------------------------------------------------------------------------
 Bitmap Heap Scan on children  (cost=4.24..14.91 rows=11 width=8)
   Recheck Cond: (parent_id = 999)
   ->  Bitmap Index Scan on children_parent_id_idx  (cost=0.00..4.24 rows=11 width=0)
         Index Cond: (parent_id = 999)
(4 rows)

KafkaをDockerで起動してRubyスクリプトからメッセージを送受信する

Sat, 21 Mar 2020 00:00:00 -0500

概要

とりあえずKafkaをDockerで立ち上げてローカルのRubyスクリプトから接続する手順を紹介します。実際にはKafkaクラスタの構成や設定、その他Kafkaをプログラムから使う上で把握しておいた方が良いことが多々ありますが、全て割愛します。

前提ソフトウェア

ソフトウェア	バージョン	備考
bitnami/kafka	2.4.1-debian-10-r8	Dockerイメージ
bitnami/zookeeper	3.6.0-debian-10-r12	Dockerイメージ
ruby	2.6.3p62	-
ruby-kafka	1.0.0	-

準備

docker-compose.yml

Kafkaが動作するにはZooKeeperも必要なのでこのエントリではdocker-composeを使います。Kafkaのイメージはbitnamiが提供しているDockerイメージを使います。KafkaのDockerイメージとしてはDL数も10M+で最もメジャーなものかと思います。

https://hub.docker.com/r/bitnami/kafka

以下の内容をdocker-compose.ymlに記述して下さい。この内容はbitnamiが提供するdocker-compose.ymlに対してDocker Hubに書いてあるAccessing Kafka with internal and external clientsの変更を加えたものになります。

DockerボリュームにZooKeeper, Kafkaのデータを永続化していますので手順をゼロからやり直す場合はご注意下さい。プログラムからはローカルホストのポート29092番でKafkaブローカーに接続できるように設定しています。

version: '2'

services:
  zookeeper:
    image: 'bitnami/zookeeper:3'
    ports:
      - '2181:2181'
    volumes:
      - 'zookeeper_data:/bitnami'
    environment:
      - ALLOW_ANONYMOUS_LOGIN=yes
  kafka:
    image: 'bitnami/kafka:2'
    ports:
      - '9092:9092'
      - '29092:29092'
    volumes:
      - 'kafka_data:/bitnami'
    environment:
      - KAFKA_CFG_ZOOKEEPER_CONNECT=zookeeper:2181
      - ALLOW_PLAINTEXT_LISTENER=yes
      - KAFKA_CFG_LISTENER_SECURITY_PROTOCOL_MAP=PLAINTEXT:PLAINTEXT,PLAINTEXT_HOST:PLAINTEXT
      - KAFKA_CFG_LISTENERS=PLAINTEXT://:9092,PLAINTEXT_HOST://:29092
      - KAFKA_CFG_ADVERTISED_LISTENERS=PLAINTEXT://kafka:9092,PLAINTEXT_HOST://localhost:29092
    depends_on:
      - zookeeper

volumes:
  zookeeper_data:
    driver: local
  kafka_data:
    driver: local

Rubyスクリプト

Rubyのスクリプトを用意します。Kafkaクライアントのラッパーはいくつかありますが仕組みがわからないとドはまりするので最初は素のruby-kafkaを使うことを強くおすすめします。

https://github.com/zendesk/ruby-kafka

Gemfile

Gemfileを作ります。

bundle init

Gemfileを編集して以下の1行を追加します。

gem "ruby-kafka"

Gemをインストールします。

bundle install

メッセージ送信スクリプト

produce.rbとして以下の内容を記述します。何も指定せずに接続して、時刻が入ったメッセージをtest_topicに1回送るだけです。余計なことは一切しません。

require "kafka"

kafka = Kafka.new("127.0.0.1:29092")

kafka.deliver_message("Hello at #{Time.now.to_s}", topic: "test_topic")

メッセージ受信スクリプト

consume.rbとして以下の内容を記述します。test_topicからメッセージを読み込み続けます。このスクリプトはブロッキングするので終了しません。

require "kafka"

kafka = Kafka.new("127.0.0.1:29092")

kafka.each_message(topic: "test_topic") do |message|
  puts "offset: #{message.offset}"
  puts "key: #{message.key}"
  puts "value: #{message.value}"
end

以上で準備が整いました。

動作

Kafkaの起動

docker-composeでZooKeeperとKafkaブローカーを起動します。起動には少し時間がかかりますのでログ出力が落ち着くまで待ちます。

docker-compose up

[2020-03-21 04:49:01,561] INFO [KafkaServer id=1001] started (kafka.server.KafkaServer) というようなメッセージが出たら起動は終わっています。

Kafkaにメッセージを送信

Kafkaにメッセージを送信します。

bundle exec ruby produce.rb

Kafka::LeaderNotAvailableエラーが出るかも知れませんが無視します。不安ならもう1回実行するとエラーなく終了するはずです。

Kafkaからメッセージを受信

Kafkaからメッセージを受信します。過去に送ったメッセージがすべて出力されます。

bundle exec ruby consume.rb

ここでもう1度prodduce.rbでメッセージを送ってみると、起動中のconsume.rbで同時にメッセージが受信されることが確認できます。

Marshal.loadでクラスが存在しない場合はconst_missingはコールバックされない

Sat, 14 Mar 2020 00:00:00 -0500

概要

Marshal.load でクラスが存在しない場合は const_missing はコールバックされません。この挙動はconst_missingを利用した仕組み(例えばRailsのオートロード)で思わぬ落とし穴になることがあります。

前提ソフトウェア

ソフトウェア	バージョン	備考
Ruby	2.6.3p62	-

const_missingとMarshal.load

RubyのModule#const_missingはスクリプトの実行中に参照した定数が定義されていない時にコールバックされます。const_missingをオーバーライドすることで任意の処理を実行することができます。

ところがconst_missingはMarshal.loadで未定義のクラスが現れた時はコールバックされません。以下は検証コードです。

class Hoge; end

open('hoge.marshal', 'w') do |f|
  p hoge = Hoge.new #=> #<Hoge:0x00007f8f34116de8>

  # MarshalでHogeクラスをシリアライズして書き出し
  f.puts Marshal.dump(Hoge.new)
end

class Hoge; end

# もちろんHogeが定義されていればMarshalでデシリアライズできる
p Marshal.load(File.read('hoge.marshal')) #=> #<Hoge:0x00007f8f34116de8> 

class Module
  def const_missing(id)
    if id == :Hoge
      eval "class Hoge; end", Object::TOPLEVEL_BINDING

      Hoge
    end
  end
end

# const_missingしておけば本来はconst_missingが呼ばれる
p Hoge.new #=> #<Hoge:0x0000555de1e9fd60>

class Module
  def const_missing(id)
    if id == :Hoge
      eval "class Hoge; end", Object::TOPLEVEL_BINDING

      Hoge
    end
  end
end

# ところがMarshal.loadではconst_missingが呼ばれず即座にArgumentErrorになる
Marshal.load(File.read('hoge.marshal')) #=> ArgumentError

少し調べたところRubyにはこの件に関連する以下のissuesがあるようです。

Railsのオートロードと解決策

Railsのオートロードはconst_missingを利用した仕組みの一つです。Railsのオートロードではクラス(定数)が定義されていない時にautoload_paths以下のファイルを自動でロードするようになっています。

https://github.com/rails/rails/blob/v6.0.2/activesupport/lib/active_support/dependencies.rb#L212-L215

Rails5以降のproductionモードではeager_load = trueがデフォルトなのでなかなか起こりませんが、developmentモードで開発中の場合ならオートロードがうまくいかず問題になることがあるかも知れません。そのような場合は問題となるクラスを先にrequireするかRails.application.earger_load!してしまうのが良いでしょう。

暗黙的にMarshal.loadするGem

明示的にMarshalを使っていない場合でも内部でMarshal.dump, Marshal.loadするGemでこの問題に遭遇することがあります。

例えばParallelはプロセスによる並列化を行う場合フォークしたプロセスでMarshal.dumpして親プロセスでMarshal.loadする挙動をします。(Parallelについての詳しい説明は Rubyで並列処理を行うparallel gemの使い方と勘所をご覧下さい)

このような場合はフォーク先のプロセスでオートロードされたクラスが親プロセスに渡る時、親プロセスでは未ロードでconst_missingもコールバックされないため例外が発生します。

leveldb-rubyでRubyからLevelDBを使う

Sat, 07 Mar 2020 00:00:00 -0600

概要

LevelDBはC++で書かれたKVSのライブラリです。イメージとしてはSQLite3のKVS版のようなものです。LevelDBに書き込んだデータはファイルに永続化されます。leveldb-rubyはLevelDBのRubyバインディングです。leveldb-rubyを使うとLevelDBをHash風のインタフェースで扱うことができます。

前提ソフトウェア

ソフトウェア	バージョン	備考
Ubuntu	18.04	-
Ruby	2.5.1p57	-
leveldb-ruby	0.15	-

インストール

前提としてLevelDBが必要です。Debian, Ubuntuの場合はlibleveldb-devをインストールして下さい。またネイティブライブラリをビルドできる必要がありますのでそこは別途構築して下さい。

apt-get install libleveldb-dev

インストールするGemはleveldb-rubyです。

gem install leveldb-ruby

もしくはGemfileに以下を記述してbundle installして下さい。

gem "leveldb-ruby"

主な使い方と特徴

LevelDB::DB.newでデータを永続化するディレクトリを指定してLevelDBを開きオブジェクトを作ります。このオブジェクトはHash風のインタフェースで使うことができます。ただしキーと値は文字列(String)である必要があります。LevelDBはバイナリセーフであるためバイナリの文字列でもキーや値として使用できます。

require 'leveldb'

db = LevelDB::DB.new('leveldb_data')

# Hash風に読み書き
p db['foo'] = 'bar' #=> 'bar'
p db['foo'] #=> 'bar'

# deleteで削除
p db.delete('foo') #=> 'bar'
p db.delete('foo') #=> nil

# put, getを使った読み書き
p db.put('foo', 'bar') #=> 'bar'
p db.get('foo') #=> 'bar'

# バイナリのキーや値を扱える
require 'securerandom'

p bin_key = SecureRandom.random_bytes #=> "O\xF3\xA6\xD9\x966F|\xC0\xFC2\tFe}\x9E"
p bin_val = SecureRandom.random_bytes #=> "\x19\xE0\x92\x8Bs\xAE\xC1\xF2d\xCE\xBE\xE1\xD8i0\xBC"

p db[bin_key] = bin_val #=> "\x19\xE0\x92\x8Bs\xAE\xC1\xF2d\xCE\xBE\xE1\xD8i0\xBC"
p db[bin_key] #=> "\x19\xE0\x92\x8Bs\xAE\xC1\xF2d\xCE\xBE\xE1\xD8i0\xBC"

シェルからファイルを見てみればわかりますが、LevelDBのデータは指定したディレクトリ以下に永続化されます。ディレクトリやファイルは勝手に作られデータ量に応じて増えていきます。

$ find leveldb_data
leveldb_data
leveldb_data/LOCK
leveldb_data/MANIFEST-000002
leveldb_data/LOG
leveldb_data/000003.log
leveldb_data/CURRENT

LevelDB::DBはおおよそHashのように使用できますが、fetchなど一部のメソッドは利用できません。メソッドの差分は以下の通りでした。またeachやmapできる順序がキーの昇順であるなど動作には細かい違いがあります。

require 'leveldb'

db = LevelDB::DB.new('leveldb_data')

# LevelDB::DBで利用できるメソッド
p db.methods #=> [:get, :close!, :iterator, :close, :keys, :delete, :member?, :inspect, :exists?, :[], :[]=, :contains?, :includes?, :put, :size, :each, :batch, :pathname, :options, :values, :chain, :to_set, :lazy, :to_h, :include?, :max, :min, :find, :to_a, :entries, :sort, :sort_by, :grep, :grep_v, :count, :detect, :find_index, :find_all, :select, :filter, :reject, :collect, :map, :flat_map, :collect_concat, :inject, :reduce, :partition, :group_by, :first, :all?, :any?, :one?, :none?, :minmax, :min_by, :max_by, :minmax_by, :each_with_index, :reverse_each, :each_entry, :each_slice, :each_cons, :each_with_object, :zip, :take, :take_while, :drop, :drop_while, :cycle, :chunk, :slice_before, :slice_after, :slice_when, :chunk_while, :sum, :uniq, :instance_variable_defined?, :remove_instance_variable, :instance_of?, :kind_of?, :is_a?, :tap, :instance_variable_set, :protected_methods, :instance_variables, :instance_variable_get, :private_methods, :public_methods, :public_send, :method, :public_method, :singleton_method, :define_singleton_method, :extend, :to_enum, :enum_for, :<=>, :===, :=~, :!~, :eql?, :respond_to?, :gem, :freeze, :object_id, :send, :to_s, :display, :nil?, :hash, :class, :singleton_class, :clone, :dup, :itself, :yield_self, :then, :taint, :tainted?, :untaint, :untrust, :untrusted?, :trust, :frozen?, :methods, :singleton_methods, :equal?, :!, :==, :instance_exec, :!=, :instance_eval, :__id__, :__send__]

# HashにあってLevelDB::DBにないメソッド
p ({}.methods - db.methods) #=> [:index, :<=, :replace, :clear, :>=, :empty?, :fetch, :shift, :select!, :length, :values_at, :filter!, :delete_if, :reject!, :keep_if, :assoc, :rassoc, :compact, :flatten, :>, :compact!, :to_hash, :to_proc, :<, :default, :rehash, :store, :default=, :default_proc, :default_proc=, :each_value, :each_key, :each_pair, :transform_keys, :transform_keys!, :transform_values, :transform_values!, :fetch_values, :update, :slice, :merge!, :invert, :has_key?, :has_value?, :merge, :key?, :value?, :compare_by_identity, :compare_by_identity?, :dig, :key]

ハッシュにない機能として範囲指定によるアクセスが可能です。LevelDBの内部はLSTが用いられており、データはキーでソートされた状態で保存されます。このためLevelDBは範囲指定でキーと値を取得することができます。以下は範囲指定でeachする例です。

require 'leveldb'

db = LevelDB::DB.new('leveldb_data')

db['001'] = 'a'
db['002'] = 'b'
db['003'] = 'c'
db['004'] = 'd'
db['005'] = 'e'

db.each(from: '002', to: '004') do |k, v|
  p k, v #=> "002" "b" "003" "c" "004" "d"
end

LevelDBのデータはデフォルトでSnappyで透過的に圧縮されます。Snappyによる圧縮がどの程度うまくいくかによりますが、永続化に要する容量はプログラムで扱う際の容量よりも小さくなると期待できます。

LevelDBにはいくつかオプションがあり LevelDB::DB.newの第二引数で指定することができます。Snappyによる圧縮有無の他にブロックサイズやバッファサイズ等を調節することができます。leveldb-rubyで指定できるオプションは Class: LevelDB::Options にまとまっています。以下は圧縮無しに指定する例です。既に作成済みのDBを別のオプションで開くとエラーとなることがあります。

db = LevelDB::DB.new 'leveldb_data', compression: LevelDB::CompressionType::NoCompression
p db.options #=> #<LevelDB::Options:0x0000556b63952c60 @create_if_missing=true, @error_if_exists=false, @paranoid_checks=false, @write_buffer_size=4194304, @max_open_files=1000, @block_size=4096, @block_restart_interval=16, @compression=LevelDB::CompressionType::NoCompression>

性能

LevelDB事態の性能については少々古いですがGoogleによるベンチマークがありますので以下を参照して下さい。SQLite3, Kyoto Cabinet’sと性能を比較しLevelDBが高速だとしています。

LevelDB Benchmarks

雑ですがleveldb-ruby経由で性能を確認してみます。検証環境はAWSのt3.nano(2CPU, 512MB RAM)に30GBのEBSボリュームを利用しています。T2/T3 Unlimitedは有効で、EBSのバーストクレジットは十分な状態です。実行するプログラムは以下のとおりです。念のためページキャッシュをクリアしています。値を作る処理もありますので純粋なleveldb-rubyの性能ではなく参考程度となります。

#!/bin/bash
rm -r leveldb_data

sudo sh -c 'echo 1 > /proc/sys/vm/drop_caches'
bundle exec ruby write.rb

sudo sh -c 'echo 1 > /proc/sys/vm/drop_caches'
bundle exec ruby read.rb

# read.rb
require 'leveldb'
require 'benchmark'

db = LevelDB::DB.new('leveldb_data')

Benchmark.bm do |x|
  x.report "Read" do
    10_000_000.times do |i|
      k = "%016d" % i

      db[k]
    end
  end
end

# write.rb
require 'leveldb'
require 'benchmark'

db = LevelDB::DB.new('leveldb_data')

Benchmark.bm do |x|
  x.report "Write" do
    10_000_000.times do |i|
      k = v = "%016d" % i

      db[k] = v
    end
  end
end

実行結果は以下のとおりです。1,000万回の書き込みに30秒、1,000万回の読み込みに16秒でした。

       user     system      total        real
Write 19.884539  12.670534  32.555073 ( 30.172090)
       user     system      total        real
Read 15.850135   0.020184  15.870319 ( 16.018496)

データをファイルに永続化するためLevelDBの読み書きではIOが発生します。書き込みはデフォルトでバッファリングされます。実際の性能はLevelDBに指定するオプション、OSのページキャッシュ、IO性能に依存します。用途にあわせて計測されることをおすすめします。

想像される用途

ぱっと想像されるのはRubyでハッシュのデータを永続化したいとき、もしくは永続化したデータを再利用したい場合でしょうか。例えばプログラムに対する設定値や辞書のようなデータをLevelDBに保存しておくような使い方が考えられます。

範囲でデータを取得できることに注目した用途もありそうです。例えばキーにタイムスタンプを含めて時系列データを格納し範囲指定で参照するような使い方も考えられます。

実メモリに収まらない大きなデータをハッシュのように扱いたい場合も利用できます。ただしメモリが足りなくなると頻繁にIOが発生して遅くなるため、とりあえず動くようにはできますが、実用的な速度で使えるかは用途次第かと思います。

注意点

並列アクセス

LevelDBの制限で、同じLevelDBを開けるプロセスは1つです。マルチスレッドで読み書きすることはできますが、マルチプロセスで読み書きすることはできません。複数のプロセスで同時に同じLevelDBを開くとIO error: lock leveldb_data/LOCK: Resource temporarily unavailable (LevelDB::Error)が発生します。

leveldb-rubyのメンテナンス

これまでleveldb-rubyを紹介してきましたが、このライブラリは残念ながらあまりメンテされていません。Gemに同梱されているLevelDBのバージョンは8年前のものです。LevelDB 1.2に対応するPRも出ていますが放置されています。

LevelDB up to v1.20 by muyesh · Pull Request #40 · wmorgan/leveldb-ruby

LevelDBのRubyバインディングはleveldb-rubyはダウンロード数から最もメジャーだと考えています。他にLevelDBをRubyから扱えるようにするライブラリにはFFI経由でLevelDBを使うleveldbやLevelDBを同梱しないネイティブなバインディングのleveldb-nativeがあります。

アプリケーションでの利用

ファイルにデータを保存するKVSのため複数サーバ、複数プロセスで動作させることが前提になっているWebアプリケーションやワーカーにはあまり適さないかも知れません。

そもそも主キーによる参照やカバリングインデックスによる範囲アクセスであればRDBMSでも十分高速であることが多いためKVSが必要かも一考しましょう。本当にKVSが必要ということであればまずはRedisやAWSのDynamoDBなど一般的なデータストアの使用を検討しましょう。

特にLevelDBによる永続化を目的とする場合はデータの可搬性やバックアップについて考慮が必要です。カジュアルに使い始めることはできますが、実際にLevelDBをどのように活用するかはいろいろ考える必要があります。

利用例

拙作ですが具体的な利用例を紹介します。BestGems.orgではGemのダウンロード数やランキングの推移のデータを保存するのにleveldb-rubyを使っています。他のデータ保持方法も検討しましたが、t3.micro(2CPU, 1GB RAM)にアプリケーション本体とPostgreSQLが同居するタイトな動作環境で、100tps程度のAPI呼び出しに対して十分な処理性能を得られたのはLevelDBだけでした。

データの量は現時点でおよそ1,000万キー、圧縮後の容量にして4GBほどです。LevelDBにアクセスする専用のプロセスを用意しアプリケーションサーバからdRuby経由でRPCでアクセスするようにしています。これまで2年以上これといって問題にならず利用ができています。

AWS SDK for Rubyでhttp_wire_traceオプションを使ってAPIとの通信内容を出力する

Mon, 17 Sep 2018 00:00:00 -0500

概要

AWS SDK for Rubyではhttp_wire_traceオプションを利用するとAWS SDKがAPIと通信した内容を出力することができます。デバッグ時に便利です。
http_wire_traceオプションはリソースやクライアントの初期化時に指定するかAWS.configでグローバルに指定することもできます。
使い方は言語毎に異なりますが、少なくともPythonのboto3でも通信内容の出力は可能で、他の言語のSDKにも同様の機能があるようです。

前提ソフトウェア・ハードウェア

ソフトウェア	バージョン	備考
Ruby	2.5.1p57	-
aws-sdk-core	3.27.0	-
aws-sdk-sns	1.5.0	-

Amazon SNSのpublishで通信内容を出力する例

AWS SDKを使ってコードを書いているとAWSのAPIとの通信内容を確認してデバッグしたくなることがあります。そのような時はhttp_wire_traceオプションを使うと便利です。

通信内容を出力しないサンプルコード

とあるAmazon SNSのトピックにPublishアクションを実行する以下のコードを例に通信内容を出力してみます。このコードを実行すると、通常は何も出力されずに正常終了します。

require 'aws-sdk-sns'

topic = Aws::SNS::Topic.new(arn: 'arn:aws:sns:ap-northeast-1:9999999999999:some_topic')

topic.publish(message: 'Hello!')

通信内容を出力するよう改変したサンプルコード

ここでAws::SNS::Topicの初期化時にhttp_wire_trace: trueオプションを指定してみます。これでtopicに代入されたAws::SNS::Topicインスタンスで行った操作は通信内容が出力されるようになります。

require 'aws-sdk-sns'

topic = Aws::SNS::Topic.new(
  arn: 'arn:aws:sns:ap-northeast-1:9999999999999:some_topic',
  http_wire_trace: true # http_wire_trace: true を追加した
)

topic.publish(message: 'Hello!')

プログラムは正常終了しますが、以下の標準出力がされるようになります。クレデンシャルなど機微な情報が含まれることに注意が必要です。

opening connection to sns.ap-northeast-1.amazonaws.com:443...
opened
starting SSL for sns.ap-northeast-1.amazonaws.com:443...
SSL established
<- "POST / HTTP/1.1\r\nContent-Type: application/x-www-form-urlencoded; charset=utf-8\r\nAccept-Encoding: \r\nUser-Agent: aws-sdk-ruby3/3.27.0 ruby/2.5.1 x86_64-linux-gnu aws-sdk-sns/1.5.0\r\nHost: sns.ap-northeast-1.amazonaws.com\r\nX-Amz-Date: 20180917T114946Z\r\nX-Amz-Content-Sha256: XXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXX\r\nAuthorization: AWS4-HMAC-SHA256 Credential=XXXXXXXXXXXXXXXXXXXX/20180917/ap-northeast-1/sns/aws4_request, SignedHeaders=content-type;host;x-amz-content-sha256;x-amzignature=XXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXX\r\nContent-Length: 120\r\nAccept: */*\r\n\r\n"
-> "HTTP/1.1 200 OK\r\n"
-> "x-amzn-RequestId: XXXXXXXX-XXXX-XXXX-XXXXXXXXXXXXXXXXX\r\n"
-> "Content-Type: text/xml\r\n"
-> "Content-Length: 294\r\n"
-> "Date: Mon, 17 Sep 2018 11:49:46 GMT\r\n"
-> "\r\n"
reading 294 bytes...
-> ""
-> "<PublishResponse xmlns=\"http://sns.amazonaws.com/doc/2010-03-31/\">\n  <PublishResult>\n    <MessageId>XXXXXXXX-XXXX-XXXX-XXXX-XXXXXXXXXXXX</MessageId>\n  </PublishResult>\n  <ResponseMetadata>\n    <RequestId>XXXXXXXX-XXXX-XXXX-XXXXXXXXXXXXXXXXX</RequestId>\n  </ResponseMetadata>\n</PublishResponse>\n"
read 294 bytes
Conn keep-alive

同様の出力をする他のサンプルコード

クライアントやリソースのインスタンスに設定したhttp_wire_traceは受け継がれていきます。以下の2つの例ではAws::SNS::Topicインスタンスそのものにはオプションを指定していませんがtopic.publishで通信内容が出力されます。

require 'aws-sdk-sns'

sns = Aws::SNS::Resource.new(http_wire_trace: true) # Aws::SNS::Resourceでhttp_wire_trace: trueを指定

topic = sns.topic('arn:aws:sns:ap-northeast-1:9999999999999:some_topic') # Aws::SNS::ResourceからTopicを作成

topic.publish(message: 'Hello!')

require 'aws-sdk-sns'

sns_client = Aws::SNS::Client.new(http_wire_trace: true) # Aws::SNS::Clientでhttp_wire_trace: trueを指定

topic = Aws::SNS::Topic.new(arn: 'arn:aws:sns:ap-northeast-1:9999999999999:some_topic', client: sns_client) # Aws::SNS::Clientを渡してTopicを初期化

topic.publish(message: 'Hello!')

もし全てのAWSとの通信部分で出力したい場合はAws.configでhttp_wire_traceを指定すれば良いです。

require 'aws-sdk-sns'

Aws.config.update(http_wire_trace: true)

topic = Aws::SNS::Topic.new(arn: 'arn:aws:sns:ap-northeast-1:9999999999999:some_topic')

topic.publish(message: 'Hello!')

なおloggerオプションを指定するとhttp_wire_traceの出力もつられてloggerの出力先に出るようになっています。以下の例では標準エラー出力に通信内容が出力されます。

require 'aws-sdk-sns'

Aws.config.update(logger: Logger.new(STDERR), http_wire_trace: true) # loggerオプションに標準エラー出力向きのLoggerを指定している

topic = Aws::SNS::Topic.new(
  arn: 'arn:aws:sns:ap-northeast-1:9999999999999:some_topic'
)

topic.publish(message: 'Hello!')

AWS SDK for Rubyの`http_wire_trace`はどのように実装されているのか?

http_wire_traceオプションによって出力をするかどうか切り替えている箇所はSeahorse::Client::NetHTTP::ConnectionPoolの以下の行です。

aws-sdk-ruby/connection_pool.rb at 2080dae8a525cf87ade12c65d1e06d3ee3c4ad26 · aws/aws-sdk-ruby

 http.set_debug_output(logger) if http_wire_trace?

このコード中でhttpはExtendedSessionクラス(RubyのNet::HTTPのデリゲータ)のインスタンスです。 http_wire_traceとはAWS SDK内部でNet::HTTPのset_debug_outputにloggerを渡すかどうかのフラグであることがわかります。

Net::HTTP#set_debug_outputはリファレンスを参照して下さい。内部では<<しているようです。

AWS SDKにloggerを設定するとhttp_wire_traceの出力先も変更されるのはこのためです。ただし<<で出力しているのでRubyのLoggerの実装ではフォーマットはされません。

instance method Net::HTTP#set_debug_output (Ruby 2.5.0)

set_debug_output(io) -> ()

デバッグ出力の出力先を指定します。このメソッドは深刻なセキュリティホールの原因になるため、デバッグ以外では決して使わないでください。

io に nil を指定するとデバッグ出力を止めます。

[PARAM] io:

出力先を指定します。このオブジェクトはメソッド « を持っている必要があります。

他の言語のAWS SDKで同様の機能はあるか

Pythonのboto3だとboto3.set_stream_loggerで出力可能でした。 RubyのコードをPythonで書き換えたサンプルコードは以下のとおりです。

Boto3 Reference – Boto 3 Docs 1.9.4 documentation

import boto3

boto3.set_stream_logger(name='')

sns = boto3.resource('sns')

topic = sns.Topic('arn:aws:sns:ap-northeast-1:999999999999:some_topic')

topic.publish(Message='Hello!')

参考

ChromebookでLinuxを動かす。Chrome OS 69でC101PAは完璧なLinuxノートになる。

Sun, 02 Sep 2018 00:00:00 -0500

概要

C101PAをChrome OS 69にアップデートしLinux(Debian stretch)が動作させられることを確認しました。Chrome OS 69は現在はBetaチャンネルでアップデートできます。
Linuxを動かすのに開発者モードにする必要はありません。CromebrewやCroutonその他トリッキーな方法は不要です。
デフォルトでGUIが使えます。クリップボードもChrome OSと共有されます。ホームディレクトリ以下のファイルはChrome OSから読み書きできます。概ね良好な使い勝手です。

前提ソフトウェア・ハードウェア

ソフトウェア/ハードウェア	バージョン	備考
C101PA	-	-
Chrome OS	69.0.3497.73	Betaチャンネル

セットアップ方法

セットアップは簡単でした。

この手順でセットアップするのはProject Crostiniと呼ばれているものです。英語でぐぐる場合はProject Crostiniでぐぐると良いです。

Chrome OSのチャンネルをBetaチャンネルに変更

まだStableチャンネルでは提供されていないので、Chrome OS 69が利用できるBetaチャンネルを使いました。厳密には8月15日にBetaチャンネルで提供されたバージョンから利用可能になったようです。じきにStableチャンネルも69系になるでしょうから、そうなったらこの手順は不要になります。

切り替え方法は公式のChromebookヘルプに日本語化された解説があります。ご自身で試される場合は、注意点を良く読んで、この手順に従って下さい。

Stable、Beta、Dev チャンネルを切り替える

設定を開きLinuxをオンにする

アップデート後に設定を開くと「Linux(ベータ版)」という項目が表れます。「オンにする」ボタンが表示されているのでそこをクリックします。

Linuxをインストールする

ダイアログが出てくるので「インストール」を押します。インストールがはじまります。10分くらい待ちました。

ターミナルが立ち上がる

Linuxがインストールされてターミナルが立ち上がります。 (プロンプトにユーザ名が表示されるので、このスクリーンショットは設定を変更しています)

あとは好きなようにLinuxが使えます。はじめて使った人が気になるだろうところは以下です。

ディストリビューションはDebian stretchです。
Googleアカウントのユーザ名でユーザが作成されます。
パスワードなしでsudoが使えます。
TerminalというChromeアプリケーションでLinuxのシェルを起動できるようになります。

以上でセットアップは終了です。 OSの機能となったため、chromebrewやcroutonよりだいぶ簡単になりました。

使い勝手など

GUI

このスクリーンショットはsudo apt-get install x11-appsしてxeyesを立ち上げた状態のものです。

DISPLAY, WAYLAND_DISPLAY環境変数が設定されており、何も考えずともディスプレイサーバが提供され、GUIアプリケーションが起動できます。ウィンドウマネージャはChrome OSに統合されたものが利用できます。 Chromeや他のChromeアプリケーションとLinuxのアプリケーションはシームレスに表示され操作できます。

クリップボード

動作を見ている限りセレクション(いわゆるクリップボード)はCLIPBOARDのみがChrome OSと共有されるようです。つまりChrome OS側でコピーすると、そのままLinuxアプリケーションにペーストできます。逆もまた然りです。

文字列を選択して中クリックでペーストするPRIMARYのセレクションはChrome OSと共有できないようです。

日本語入力

Chrome OSの日本語入力はLinuxアプリケーションで利用できないようです。

逆にLinuxにインプットメソッドと日本語入力システムをインストールすればLinuxで日本語入力できるようになります。このようにインストールしたインプットメソッドはChrome OS側には利きません。

下手に気を利かされるより、お気に入りの方法で入力できた方が不便がないでしょうから、これが良いと思います。ただしChrome OS側の日本語入力がオンの状態だと、Linuxアプリケーションに正確にキー入力がされないようで注意が必要です。

ファイル

Chrome OSとLinxuの間でファイルをやり取りできます。具体的にはLinuxのホームディレクトリ配下が「Linuxファイル」としてChrome OSのファイラで見えるようになり、Chrome OSから読み書きできます。とても便利です。

なおdebパッケージをChrome OSのファイラからLinux内にインストールすることもできるようです。

ランチャー

LinuxにインストールしたアプリケーションはChrome OSのランチャーに表示されるようになります。

XTermだろうがGVimだろうがFirefoxだろうが、Linux側でメニューを作成するアプリケーションをインストールさえすれば、なんでも起動できます。

ARM

これはC101PAのハードウェアの話です。

C101PAのCPUはARMなので、ARMで動作しない、ARM用のパッケージが用意されていないLinuxアプリケーションはそのままでは使えません。こればかりは人によりますが、困る可能性が高いのはDropboxのクライアントやMozc(Google日本語入力)だと思います。

私自身は出先でC101PA(+ crouton)だけという生活を9ヶ月ほど続けていましたが、もう慣れて全く問題になっていません。

まとめ

もしLinuxをノートPCで動作させようと思うと、未だにハードウェアとの相性に苦戦することはままあります。少なくともChromebookであればノートPCに備わった機能は大抵まともに動作しますから、足回りを気にせずLinuxアプリケーションが動作する環境を手に入れられるのは大きなメリットです。加えてChromebookにはC101PAのように安価で小型なモデルもあるのもうれしいところです。

ASUS Chromebook Flip C101PA シルバー 10.1型ノートPC OP1 Hexa-core/4GB/eMMC16GB/C101PA-OP1

posted with amazlet at 18.09.02

ASUSTOR (2017-12-12)
売り上げランキング: 5,177

Amazon.co.jpで詳細を見る

Chrome OSでLinuxが簡単に使えるようになったことでChromebook活用の幅が一気に広がりました。端末、エディタ、Linuxの標準的なツールが動けば幸せという人にはかなりの朗報と言えるのではないでしょうか。

Rubyで並列処理を行うparallel gemの使い方と勘所

Sun, 22 Jul 2018 00:00:00 -0500

parallelを使うとKenrel#forkやThreadを駆使するのと比べて簡単に並列処理を書くことができます。parallelは拙作のBestGems.orgによると、合計ダウンロード数で151位、デイリーダウンロード数は100位前後で、現時点で非常にメジャーなGemとなっています。

この記事ではparallelの基本的な使い方と、実際に使ってみて感じた注意点をTipsとして整理したいと思います。

parallelはREADME.mdが親切に書かれています。加えて主要な部分は500行程度の小さなGemです。利用する場合は公式のドキュメントとソースコードを確認されることをおすすめします。

前提ソフトウェア

ソフトウェア	バージョン	備考
ruby	2.5.1	-
parallel	1.12.1	-
rails	5.0	-

使い方

インストール

gem install parallelするかGemfileに以下の1行を追加してbundle installして下さい。

gem 'parallel'

明示的にロードする場合はparallelを利用するRubyのプログラムでrequire 'parallel'して下さい。以降のサンプルコードではこの記述は省略しています。

できること

parallelにはRubyのeachやmapに相当する操作を並列処理するための以下のメソッドがあります。

Rubyのメソッド	対応するparallelのメソッド
`Enumerable#each`	`Parallel.each`
`Enumerable#map`	`Parallel.map`
`Enumerable#any?`	`Parallel.any?`
`Enumerable#all?`	`Parallel.all?`
`Enumerable#each_with_index`	`Parallel.each_with_index`
`Enumerable#map`, `Enumerator#with_index`	`Parallel.map_with_index`

each

Parallel.eachはブロックが並列に実行されるeachです。並列に処理しているためブロックの実行が完了する順序はバラバラです。戻り値はParallel.eachの引数が返ります。

result = Parallel.each(1..10) do |item|
  # 普通のeachのようだがブロックは並列に実行される
  p item ** 2
end

p result

以下はこのコードの出力の例です。

map

Parallel.mapはブロックが並列に実行されるmapです。並列に処理しているためブロックの実行が完了する順序はバラバラです。戻り値はmapと同様に入力した各要素に対応した値の配列が返ります。

result = Parallel.map(1..10) do |item|
  # 普通のmapのようだがブロックは並列に実行される
  p item ** 2
end

p result

以下はこのコードの出力の例です。

1
4
9
16
25
36
49
64
100
81
[1, 4, 9, 16, 25, 36, 49, 64, 81, 100]

ワーカー番号の取得

上記のメソッド以外の機能もあります。ブロック内でParallel.worker_numberを呼ぶとワーカースレッド/プロセスの番号を取得できます。デバッグやロギングで利用できます。

require 'parallel'

result = Parallel.map(1..10) do |item|
  p [item ** 2, Parallel.worker_number]
end

p result

以下はこのプログラムの出力の例です。

[1, 0]
[4, 1]
[9, 0]
[16, 3]
[25, 4]
[81, 1]
[36, 5]
[49, 6]
[100, 1]
[64, 7]
[[1, 0], [4, 1], [9, 0], [16, 3], [25, 4], [36, 5], [49, 6], [64, 7], [81, 1], [100, 1]]

スレッドとプロセス

parallelのメソッドはオプションでプロセスで処理(in_processes)するかスレッドで処理(in_threads)するかを切り替えることができます。

プロセスで処理する場合

CRubyで何もオプションを指定しなければプロセスによる並列処理になります。並列数はparallel内部のParallel::ProcessorCount.countが返す論理コア数になります。

in_processes: 並列数を指定すると指定したプロセス数で並列化して実行できます。

Paralell.each(1..10, in_processes: 10); end

parallelの内部でforkして作られたワーカープロセスはメソッド呼び出しが完了するまで使い回されます。 1回のブロックの実行毎にforkさせたい場合はisolation: trueを指定することもできます。

Paralell.each(1..10, in_processes: 10, isolation: true); end

スレッドで処理する場合

JRubyで何もオプションを指定しなければスレッドによる並列処理になります。並列数はparallel内部のParallel::ProcessorCount.countが返す論理コア数になります。

in_threads: 並列数を指定すると指定したスレッド数で並列化して実行できます。

Paralell.each(1..10, in_threads: 10) do; end

Tips

parallelを使ってわかった勘所をまとめます。

第1引数について

eachやmapの第1引数はどんなオブジェクトを渡すかによってparallelの挙動は異なります。

基本的に第1引数はto_aされてparallelの内部で配列になります。この挙動を知らないとparallelの呼び出し元で意図せずしてメモリ使用量が増大することがあります。例外的にProcとThread::Queueを第1引数とした場合に、配列にはされずに処理します。

第一引数	parallelによる判定条件	挙動
`Proc`オブジェクト	`.call`できること	ブロックの実行ごとに`.call`します。`.call`で`Parallel::Stop`を返すと処理を完了します。
`Thread::Queue`オブジェクト	`num_waiting`および`pop`できること	ブロックの実行ごとに`.pop`します。
その他	上記以外	まず`to_a`して配列にします。ブロックの実行ごとに先頭から要素を処理します。

詳細はparallelのJobFactoryクラスのソースを見て下さい。

https://github.com/grosser/parallel/blob/v1.12.1/lib/parallel.rb#L89-L145

Rangeや他のEnumerableを渡しても動作しますが、呼び出し元でArray, Proc, Thread::Queueのいずれかを渡した方が、誤解を招かないコードになると思います。

ワーカープロセスとの通信について

parallelはワーカープロセスとの通信をIO.pipeで生成したパイプの入出力で行います。ワーカープロセスとのオブジェクトの受け渡しはMarshal.dump, Marshal.loadを使います。このためMarshalでシリアライズできないオブジェクトをワーカープロセスと受け渡すことはできません。

オプションについての注意点

実装上parallelのメソッドのオプションは、パラメータ引数にはなっておらず、キー名のチェックもされません。このためtypoしたオプションは無視されます。

例えば以下のコードは10プロセスで並列化することを意図しています。しかしin_processesを誤ってin_processとtypoしているため、デフォルトどおり論理コア数のプロセスで並列化されてしまいます。

Parallel.each(1..10, in_process: 10) do; end

parallelの呼び出し時にはオプションをtypoしないよう細心の注意を払いましょう。

ブロック内での例外の発生やreturnについて

ブロック内で例外(Parallel::Break, Parallel::Killを除く)を発生させたりreturnしたりするとparallelの呼び出しは例外を発生させます。この時にparallelの呼び出し元でrescueできる例外は、並列処理がスレッドとプロセスどちらか、発生した例外がStandarErrorのサブクラスかそれ以外か、により様々です。

以下はparallelのブロック内で何かまずいことが起こった時にparallelの呼び出し元でどのような例外が発生するかの例です。 parallelのソースを読めばなぜこうなるのかわかりますが、仕組みを理解していないと挙動を推し量ることは難しいかも知れません。

begin
  Parallel.each([1, 2, 3], in_threads: 2){ raise StandardError }
rescue Exception => e
  p e.class #=> StandardError
end

begin
  Parallel.each([1, 2, 3], in_processes: 2){ raise StandardError }
rescue Exception => e
  p e.class #=> StandardError
end

begin
  Parallel.each([1, 2, 3], in_threads: 2){ raise Exception }
rescue Exception => e
  p e.class #=> Exception
end

begin
  Parallel.each([1, 2, 3], in_processes: 2){ raise Exception }
rescue Exception => e
  p e.class #=> Parallel::DeadWorker
end

begin
  Parallel.each([1, 2, 3], in_threads: 2){ return }
rescue => e
  p e.class #=> LocalJupError
end

begin
  Parallel.each([1, 2, 3], in_processes: 2){ return }
rescue => e
  p e.class #=> Parallel::DeadWorker
end

例外以外の出力について

アプリケーションが出力するログではparallelで落ちる原因がわからない場合があるかも知れません。その場合はRubyプロセスが何か出力していないかも確認して下さい。例外やバックトレースからはわからない情報が出力されていることがあります。

`Parallel::DeadWorker`について

一番厄介なのはプロセスによる並列処理で発生するParallel::DeadWorkerです。

もし例外が発生するコードや意図せずブロックを抜ける箇所も存在しないのにParallel::DeadWorkerが発生する場合は、ワーカープロセスのメモリ使用量が増加したことでメモリ不足に陥ったことも疑って下さい。 NoMemoryError(Exceptionのサブクラス)がブロック内で発生してParallel::DeadWorkerとなっている可能性があります。

Railsでparallelを利用する

Railsでparallelを利用する場合のTipsです。

ActiveRecordのコネクションについて

parallelに限らずアプリケーションのコードでプロセスやスレッドを作ってActiveRecordを使う際にはコネクションをケアしなければならない時があります。 ActiveRecordのコネクションプールには明るくないため詳しくは説明しません。対処法はparallelのREADME.mdに記載がありますので参考にして下さい。

# reproducibly fixes things (spec/cases/map_with_ar.rb)
Parallel.each(User.all, in_processes: 8) do |user|
  user.update_attribute(:some_attribute, some_value)
end
User.connection.reconnect!

# maybe helps: explicitly use connection pool
Parallel.each(User.all, in_threads: 8) do |user|
  ActiveRecord::Base.connection_pool.with_connection do
    user.update_attribute(:some_attribute, some_value)
  end
end

# maybe helps: reconnect once inside every fork
Parallel.each(User.all, in_processes: 8) do |user|
  @reconnected ||= User.connection.reconnect! || true
  user.update_attribute(:some_attribute, some_value)
end

どうやってparallelを使うかによっても様々だと思いますが、基本的にはブロック内およびparallelの呼び出し直後でreconnect!でコネクションを取得し直すようにすれば、ActiveRecord絡みのエラーは起こらなくなるはずです。

大きなテーブルの中身を並列処理したい場合

parallelは使っていますが普通のRailsアプリケーションで大きなテーブルを扱う時の書き方と変わりません。 find_in_batchesやin_batchesを使ってちょっとずつテーブルから読んで処理すると良いです。

SomeModel.find_in_batches do |some_models|
  Parallel.each(some_models) do |some_model|
    # 処理
  end
end

parallelのREADME.mdにあるようにparallelのメソッドにSomeModel.allを渡す際は注意して下さい。テーブルの中身をすべて読み込んでRubyのオブジェクトとしてメモリに乗ってしまいます。

SomeModel.allの戻り値はSomeModel::ActiveRecord_Relationです。このオブジェクトは.callも.num_waiting, .popもできません。第1引数の注意点として説明したとおり、このような引数を渡すとparallelの内部でto_aされるため、テーブルの全ての内容を一気に読み込むことになります。