Docker の containerd ランタイムからイベントを受信
Docker の containerd ランタイムからイベントを受信する処理を Go 言語で実装してみました。
今回のコードは http://github.com/fits/try_samples/tree/master/blog/20201206/
はじめに
ここでは、snap でインストールした Docker の containerd へ接続します。
バージョンは以下の通りです。
$ docker version ・・・ Server: Engine: Version: 19.03.11 API version: 1.40 (minimum version 1.12) Go version: go1.13.12 Git commit: 77e06fd Built: Mon Jun 8 20:24:59 2020 OS/Arch: linux/amd64 Experimental: false containerd: Version: v1.2.13 GitCommit: 7ad184331fa3e55e52b890ea95e65ba581ae3429 ・・・
containerd へ接続するために containerd.sock を使いますが、この環境では以下のパスで参照できるようになっていました。
/var/snap/docker/current/run/docker/containerd/containerd.sock
更に、ネームスペースは docker ではなく moby となっていました。
コンテナのリストアップ
まずは、ネームスペースの確認も兼ねて containerd で実行中のコンテナをリストアップしてみます。
listup.go
package main import ( "context" "log" "github.com/containerd/containerd" "github.com/containerd/containerd/namespaces" ) const ( address = "/var/snap/docker/current/run/docker/containerd/containerd.sock" ) func main() { client, err := containerd.New(address) if err != nil { log.Fatal(err) } defer client.Close() ctx := context.Background() ns, err := client.NamespaceService().List(ctx) if err != nil { log.Printf("ERROR %v", err) return } log.Printf("namespaces: %#v", ns) // ネームスペースの設定 ctx = namespaces.WithNamespace(ctx, ns[0]) cs, err := client.Containers(ctx) if err != nil { log.Printf("ERROR %v", err) return } log.Printf("containers: %#v", cs) }
実行
ビルド
$ go build listup.go
Docker で何も実行していない状態で実行した結果です。
実行結果1
$ sudo ./listup 2020/12/06 04:16:00 namespaces: []string{"moby"} 2020/12/06 04:16:00 containers: []containerd.Container(nil)
docker run した後の実行結果は次のようになりました。
docker run 実行
$ docker run --rm -it alpine /bin/sh / #
実行結果2
$ sudo ./listup 2020/12/06 04:20:34 namespaces: []string{"moby"} 2020/12/06 04:20:34 containers: []containerd.Container{(*containerd.container)(0xc00017e180)}
Subscribe 処理1
Subscribe メソッドを呼び出して containerd からのイベントを受信してみます。
Subscribe は以下のようなシグネチャとなっています。
https://github.com/containerd/containerd/blob/master/client.go
func (c *Client) Subscribe(ctx context.Context, filters ...string) (ch <-chan *events.Envelope, errs <-chan error) { ・・・ }
events.Envelope はこのようになっています。
https://github.com/containerd/containerd/blob/master/events/events.go
type Envelope struct { Timestamp time.Time Namespace string Topic string Event *types.Any }
イベントの内容は Event フィールドに格納されており、このようになっています。
https://github.com/gogo/protobuf/blob/master/types/any.pb.go
type Any struct { TypeUrl string `protobuf:"bytes,1,opt,name=type_url,json=typeUrl,proto3" json:"type_url,omitempty"` Value []byte `protobuf:"bytes,2,opt,name=value,proto3" json:"value,omitempty"` XXX_NoUnkeyedLiteral struct{} `json:"-"` XXX_unrecognized []byte `json:"-"` XXX_sizecache int32 `json:"-"` }
TypeUrl に型名、Value に Protocol Buffers でシリアライズした結果（バイナリデータ）が設定されており、イベントの内容を参照するには typeurl.UnmarshalAny を使ってデシリアライズする必要がありました。
なお、UnmarshalAny でイベントの内容を復元するには github.com/containerd/containerd/api/events を import する必要がありました。※
※ import しておかないと
type with url containerd.TaskExit: not found のようなエラーが返される事になります
また、listup.go では Context へネームスペースを設定しましたが、ここでは containerd.WithDefaultNamespace を使ってデフォルトのネームスペースを containerd.New 時に指定するようにしています。
subscribe1.go
package main import ( "context" "log" "github.com/containerd/containerd" "github.com/containerd/containerd/events" "github.com/containerd/typeurl" // UnmarshalAny でイベントの内容を復元するために以下の import が必要 _ "github.com/containerd/containerd/api/events" ) const ( address = "/var/snap/docker/current/run/docker/containerd/containerd.sock" namespace = "moby" ) func printEnvelope(env *events.Envelope) { // イベント内容のデシリアライズ event, err := typeurl.UnmarshalAny(env.Event) if err != nil { log.Printf("ERROR unmarshal %v", err) } log.Printf( "topic = %s, namespace = %s, event.typeurl = %s, event = %v", env.Topic, env.Namespace, env.Event.TypeUrl, event, ) } func main() { client, err := containerd.New( address, containerd.WithDefaultNamespace(namespace), // デフォルトのネームスペースを指定 ) if err != nil { log.Fatal(err) } defer client.Close() ctx := context.Background() // Subscribe の実施 ch, errs := client.Subscribe(ctx) for { select { case env := <-ch: printEnvelope(env) case e := <-errs: // log.Fatal は defer を実行せずに終了してしまう log.Fatal(e) } } }
実行
ビルドと実行
$ go build subscribe1.go $ sudo ./subscribe1
docker run を実行した後、すぐに終了してみます。
docker run 実行と終了
$ docker run --rm -it alpine /bin/sh / # exit
出力結果は以下のようになりました。
subscribe1 出力結果
$ sudo ./subscribe1 2020/12/06 04:33:18 topic = /containers/create, namespace = moby, event.typeurl = containerd.events.ContainerCreate, event = &ContainerCreate{ID:c7bed・・・,Image:,Runtime:&ContainerCreate_Runtime{Name:io.containerd.runtime.v1.linux,Options:&types.Any{TypeUrl:containerd.linux.runc.RuncOptions,Value:[10 4 114 ・・・ 99],XXX_unrecognized:[],},XXX_unrecognized:[],},XXX_unrecognized:[],} 2020/12/06 04:33:18 topic = /tasks/create, namespace = moby, event.typeurl = containerd.events.TaskCreate, event = &TaskCreate{ContainerID:c7bed・・・,Bundle:/var/snap/docker/471/run/docker/containerd/daemon/io.containerd.runtime.v1.linux/moby/c7bed・・・,Rootfs:[]*Mount{},IO:&TaskIO{・・・,},Checkpoint:,Pid:5926,XXX_unrecognized:[],} 2020/12/06 04:33:18 topic = /tasks/start, namespace = moby, event.typeurl = containerd.events.TaskStart, event = &TaskStart{ContainerID:c7bed・・・,Pid:5926,XXX_unrecognized:[],} 2020/12/06 04:33:24 topic = /tasks/exit, namespace = moby, event.typeurl = containerd.events.TaskExit, event = &TaskExit{ContainerID:c7bed・・・,Pid:5926,ExitStatus:0,ExitedAt:2020-12-05 19:33:24.325472986 +0000 UTC,XXX_unrecognized:[],} 2020/12/06 04:33:24 topic = /tasks/delete, namespace = moby, event.typeurl = containerd.events.TaskDelete, event = &TaskDelete{ContainerID:c7bed・・・,Pid:5926,ExitStatus:0,ExitedAt:2020-12-05 19:33:24.325472986 +0000 UTC,ID:,XXX_unrecognized:[],} 2020/12/06 04:33:25 topic = /containers/delete, namespace = moby, event.typeurl = containerd.events.ContainerDelete, event = &ContainerDelete{ID:c7bed・・・,XXX_unrecognized:[],}
Subscribe 処理2
switch case を使ってイベントの型毎に出力内容を変更するようにしてみます。
subscribe1.go は kill したり Subscribe でエラーが発生した場合に defer で指定した client.Close() を実行しないので、ここでは Context のキャンセル機能を使って client.Close() を呼び出して終了するようにしてみました。
subscribe2.go
package main import ( "context" "fmt" "log" "os" "os/signal" "syscall" "github.com/containerd/containerd" "github.com/containerd/containerd/events" "github.com/containerd/typeurl" apievents "github.com/containerd/containerd/api/events" ) const ( address = "/var/snap/docker/current/run/docker/containerd/containerd.sock" namespace = "moby" ) func printEnvelope(env *events.Envelope) { event, err := typeurl.UnmarshalAny(env.Event) if err != nil { log.Printf("ERROR unmarshal %v", err) } var s string // 型毎の処理 switch ev := event.(type) { case *apievents.ContainerCreate: s = fmt.Sprintf("{ id = %s, image = %s }", ev.ID, ev.Image) case *apievents.ContainerDelete: s = fmt.Sprintf("{ id = %s }", ev.ID) case *apievents.TaskCreate: s = fmt.Sprintf( "{ container_id = %s, pid = %d, bundle = %s }", ev.ContainerID, ev.Pid, ev.Bundle, ) case *apievents.TaskStart: s = fmt.Sprintf( "{ container_id = %s, pid = %d }", ev.ContainerID, ev.Pid, ) case *apievents.TaskExit: s = fmt.Sprintf( "{ container_id = %s, pid = %d, exit_status = %d }", ev.ContainerID, ev.Pid, ev.ExitStatus, ) case *apievents.TaskDelete: s = fmt.Sprintf( "{ container_id = %s, pid = %d, exit_status = %d }", ev.ContainerID, ev.Pid, ev.ExitStatus, ) } log.Printf( "topic = %s, namespace = %s, event.typeurl = %s, event = %v", env.Topic, env.Namespace, env.Event.TypeUrl, s, ) } func main() { client, err := containerd.New( address, containerd.WithDefaultNamespace(namespace), ) if err != nil { log.Fatal(err) } //defer client.Close() defer func() { log.Print("close") // defer の実施を確認するためのログ出力 client.Close() }() // 終了シグナルなどの受信設定 sig := make(chan os.Signal, 1) signal.Notify(sig, syscall.SIGINT, syscall.SIGTERM) ctx, cancel := context.WithCancel(context.Background()) go func() { s := <-sig log.Printf("syscall: %v", s) // Context のキャンセルを実施 cancel() }() ch, errs := client.Subscribe(ctx) for { select { case env := <-ch: printEnvelope(env) case e := <-errs: log.Printf("ERROR %v", e) // defer を呼び出して終了するための措置 return } } }
実行
ビルドと実行
$ go build subscribe2.go $ sudo ./subscribe2
docker run 実行と終了
$ docker run --rm -it alpine /bin/sh / # exit
subscribe2 のプロセスを kill します。
subscribe2 プロセスの終了（kill）
$ sudo kill `pgrep subscribe2`
出力結果は以下のようになり、defer の実行を確認できました。
subscribe2 出力結果
$ sudo ./subscribe2 2020/12/06 04:47:13 topic = /containers/create, namespace = moby, event.typeurl = containerd.events.ContainerCreate, event = { id = 988b・・・, image = } 2020/12/06 04:47:14 topic = /tasks/create, namespace = moby, event.typeurl = containerd.events.TaskCreate, event = { container_id = 988b・・・, pid = 6136, bundle = /var/snap/docker/471/run/docker/containerd/daemon/io.containerd.runtime.v1.linux/moby/988b・・・ } 2020/12/06 04:47:14 topic = /tasks/start, namespace = moby, event.typeurl = containerd.events.TaskStart, event = { container_id = 988b・・・, pid = 6136 } 2020/12/06 04:47:19 topic = /tasks/exit, namespace = moby, event.typeurl = containerd.events.TaskExit, event = { container_id = 988b・・・, pid = 6136, exit_status = 0 } 2020/12/06 04:47:19 topic = /tasks/delete, namespace = moby, event.typeurl = containerd.events.TaskDelete, event = { container_id = 988b・・・, pid = 6136, exit_status = 0 } 2020/12/06 04:47:20 topic = /containers/delete, namespace = moby, event.typeurl = containerd.events.ContainerDelete, event = { id = 988b・・・ } 2020/12/06 04:48:18 syscall: terminated 2020/12/06 04:48:18 ERROR rpc error: code = Canceled desc = context canceled 2020/12/06 04:48:18 close
Node.js で GraphQL over gRPC 的な事をやってみる
gRPC 上で GraphQL を扱う GraphQL over gRPC 的な処理を Node.js で試しに実装してみました。
今回のコードは http://github.com/fits/try_samples/tree/master/blog/20201124/
はじめに
GraphQL はクエリ言語なので基本的に通信プロトコルには依存していません。
Web フロントエンドの用途では Apollo GraphQL が公開している GraphQL over WebSocket Protocol が有力そうですが、マイクロサービス等の用途で GraphQL を利用する事を考えると WebSocket よりも gRPC の方が適しているように思います。
- GraphQL の Query や Mutation は gRPC の Unary RPC で実現可能
- GraphQL の Subscription は gRPC の Server streaming RPC で実現可能 ※
そこで、とりあえず実装し確認してみたというのが本件の趣旨となっています。
※ GraphQL の Subscription を使わずに
gRPC の streaming RPC で代用する事も考えられる
なお、GraphQL の処理に関しては「Deno で GraphQL」の内容をベースに、gRPC は「Node.js で gRPC を試す」の静的コード生成を用いて実装しています。
Query と Mutation - sample1
まずは Subscription を除いた Query と Mutation について実装してみます。
gRPC サービス定義
gRPC のサービス定義を下記のように定義してみました。
GraphQL のクエリは文字列で扱うので型は string、結果は実質的に JSON となるので型は google.protobuf.Struct としました。
ついでに、クエリの変数も渡せるようにして型は google.protobuf.Value としています。
ここで、google.protobuf.Struct は JSON Object を Protocol Buffers で表現するための型として定義されたもので、google.protobuf.Value は JSON Value（null、文字列、数値、配列、JSON Object 等）を表現するための型です。（参照 google/protobuf/struct.proto）
proto/graphql.proto
syntax = "proto3"; import "google/protobuf/struct.proto"; package gql; message QueryRequest { string query = 1; google.protobuf.Value variables = 2; } service GraphQL { rpc Query(QueryRequest) returns (google.protobuf.Struct); }
この proto/graphql.proto から gRPC のコードを生成しておきます。
静的コード生成
> mkdir generated > npm run gen-grpc ・・・ grpc_tools_node_protoc --grpc_out=grpc_js:generated --js_out=import_style=commonjs:generated proto/*.proto
package.json の内容は以下の通りです。
package.json
{ "name": "sample1", "version": "1.0.0", "description": "", "scripts": { "gen-grpc": "grpc_tools_node_protoc --grpc_out=grpc_js:generated --js_out=import_style=commonjs:generated proto/*.proto" }, "dependencies": { "@grpc/grpc-js": "^1.2.1", "google-protobuf": "^3.14.0", "graphql": "^15.4.0", "uuid": "^8.3.1" }, "devDependencies": { "grpc-tools": "^1.10.0" } }
サーバー実装
GraphQL を扱う gRPC サーバーを実装します。
gRPC のリクエストから GraphQL のクエリやその変数の内容を取得し、graphql 関数で処理した結果をレスポンスとして返すような処理となります。
google.protobuf.Struct や Value に該当する型は google-protobuf/google/protobuf/struct_pb.js にて Struct や Value として定義されており、プレーンな JavaScript オブジェクトと相互変換するための fromJavaScript や toJavaScript メソッドが用意されています。
下記コードでは QueryRequest の variables の内容を JavaScript オブジェクトとして取得するために toJavaScript を、graphql の処理結果を Struct で返すために fromJavaScript をそれぞれ使用しています。
server.js
const grpc = require('@grpc/grpc-js') const { GraphQLService } = require('./generated/proto/graphql_grpc_pb') const { Struct } = require('google-protobuf/google/protobuf/struct_pb') const { graphql, buildSchema } = require('graphql') const { v4: uuidv4 } = require('uuid') // GraphQL スキーマ定義 const schema = buildSchema(` enum Category { Standard Extra } input CreateItem { category: Category! value: Int! } type Item { id: ID! category: Category! value: Int! } type Mutation { create(input: CreateItem!): Item } type Query { find(id: ID!): Item } `) const store = {} // スキーマ定義に応じた GraphQL 処理の実装 const root = { create: ({ input: { category, value } }) => { console.log(`*** call create: category = ${category}, value = ${value}`) const id = `item-${uuidv4()}` const item = { id, category, value } store[id] = item return item }, find: ({ id }) => { console.log(`*** call find: ${id}`) return store[id] } } const server = new grpc.Server() server.addService(GraphQLService, { async query(call, callback) { try { const query = call.request.getQuery() const variables = call.request.getVariables().toJavaScript() // GraphQL の処理 const r = await graphql(schema, query, root, {}, variables) callback(null, Struct.fromJavaScript(r)) } catch(e) { console.error(e) callback(e) } } }) server.bindAsync( '127.0.0.1:50051', grpc.ServerCredentials.createInsecure(), (err, port) => { if (err) { console.error(err) return } console.log(`start server: ${port}`) server.start() } )
クライアント実装
クライアント側は以下のようになります。
client.js
const grpc = require('@grpc/grpc-js') const { QueryRequest } = require('./generated/proto/graphql_pb') const { GraphQLClient } = require('./generated/proto/graphql_grpc_pb') const { Value } = require('google-protobuf/google/protobuf/struct_pb') const client = new GraphQLClient( '127.0.0.1:50051', grpc.credentials.createInsecure() ) const promisify = (obj, methodName) => args => new Promise((resolve, reject) => { obj[methodName](args, (err, res) => { if (err) { reject(err) } else { resolve(res) } }) }) const query = promisify(client, 'query') const createRequest = (q, v = null) => { const req = new QueryRequest() req.setQuery(q) req.setVariables(Value.fromJavaScript(v)) return req } const run = async () => { // Item の作成 const r1 = await query(createRequest(` mutation { create(input: { category: Extra, value: 123 }) { id } } `)) console.log(r1.toJavaScript()) // 存在しない Item の find const r2 = await query(createRequest(` { find(id: "a1") { id value } } `)) console.log(r2.toJavaScript()) const id = r1.toJavaScript().data.create.id // 作成した Item の find （クエリ変数の使用） const r3 = await query(createRequest( ` query findItem($id: ID!) { find(id: $id) { id category value } } `, { id } )) console.log(r3.toJavaScript()) } run().catch(err => console.error(err))
動作確認
server.js を実行しておきます。
server.js 実行
> node server.js start server: 50051
client.js を実行した結果は以下の通りで、特に問題無く動作しているようです。
client.js 実行
> node client.js { data: { create: { id: 'item-63bb7704-27b6-44ae-b955-61cbad83248d' } } } { data: { find: null } } { data: { find: { category: 'Extra', id: 'item-63bb7704-27b6-44ae-b955-61cbad83248d', value: 123 } } }
Subscription - sample2
次は Subscription の機能を追加します。
gRPC サービス定義
gRPC のサービス定義に Subscription 用のメソッドを追加し、sample1 と同様にコードを生成しておきます。
proto/graphql.proto
syntax = "proto3"; import "google/protobuf/struct.proto"; package gql; message QueryRequest { string query = 1; google.protobuf.Value variables = 2; } service GraphQL { rpc Query(QueryRequest) returns (google.protobuf.Struct); rpc Subscription(QueryRequest) returns (stream google.protobuf.Struct); }
サーバー実装
「Deno で GraphQL」では単一の Subscription を処理するだけの実装だったので、複数クライアントからの Subscription を処理するために PubSub というクラスを追加し、subscription の呼び出し毎に MessageBox を作成、（クライアントが接続中の）有効な全ての MessageBox へメッセージを配信するようにしています。
server.js
const grpc = require('@grpc/grpc-js') const { GraphQLService } = require('./generated/proto/graphql_grpc_pb') const { Struct } = require('google-protobuf/google/protobuf/struct_pb') const { graphql, buildSchema, subscribe, parse } = require('graphql') const { v4: uuidv4 } = require('uuid') // GraphQL スキーマ定義 const schema = buildSchema(` enum Category { Standard Extra } input CreateItem { category: Category! value: Int! } type Item { id: ID! category: Category! value: Int! } type Mutation { create(input: CreateItem!): Item } type Query { find(id: ID!): Item } type Subscription { created: Item } `) class MessageBox { #promises = [] #resolves = [] #appendPromise = () => this.#promises.push( new Promise(res => this.#resolves.push(res)) ) publish(msg) { if (this.#resolves.length == 0) { this.#appendPromise() } this.#resolves.shift()(msg) } [Symbol.asyncIterator]() { return { next: async () => { console.log('*** asyncIterator next') if (this.#promises.length == 0) { this.#appendPromise() } const value = await this.#promises.shift() return { value, done: false } } } } } // クライアント毎の MessageBox を管理 class PubSub { #subscribes = [] publish(msg) { this.#subscribes.forEach(s => s.publish(msg)) } subscribe() { const sub = new MessageBox() this.#subscribes.push(sub) return sub } unsubscribe(sub) { this.#subscribes = this.#subscribes.filter(s => s != sub) } } const store = {} const pubsub = new PubSub() const root = { create: ({ input: { category, value } }) => { ・・・ }, find: ({ id }) => { ・・・ } } const server = new grpc.Server() server.addService(GraphQLService, { async query(call, callback) { ・・・ }, async subscription(call) { console.log('*** subscribed') try { const query = call.request.getQuery() const variables = call.request.getVariables().toJavaScript() const sub = pubsub.subscribe() call.on('cancelled', () => { console.log('*** unsubscribed') pubsub.unsubscribe(sub) }) const subRoot = { created: () => sub } // GraphQL の Subscription 処理 const aiter = await subscribe(schema, parse(query), subRoot, {}, variables) for await (const r of aiter) { // メッセージの配信 call.write(Struct.fromJavaScript(r)) } } catch(e) { console.error(e) call.destroy(e) } } }) server.bindAsync( ・・・ )
Subscription 用クライアント実装
Subscription を呼び出すクライアントは以下のようになります。
client_subscribe.js
const grpc = require('@grpc/grpc-js') const { QueryRequest } = require('./generated/proto/graphql_pb') const { GraphQLClient } = require('./generated/proto/graphql_grpc_pb') const { Value } = require('google-protobuf/google/protobuf/struct_pb') const client = new GraphQLClient( '127.0.0.1:50051', grpc.credentials.createInsecure() ) const req = new QueryRequest() req.setQuery(` subscription { created { id category } } `) req.setVariables(Value.fromJavaScript(null)) const stream = client.subscription(req) stream.on('data', msg => { const event = msg.toJavaScript() console.log(`*** received event = ${JSON.stringify(event)}`) }) stream.on('end', () => console.log('*** stream end')) stream.on('error', err => console.log(`*** stream error: ${err}`))
動作確認
server.js を実行した後、client_subscribe.js を 2つ実行して sample1 で作成した client.js を実行すると以下のようになりました。
server.js の出力結果
> node server.js start server: 50051 *** subscribed *** asyncIterator next *** subscribed *** asyncIterator next *** call create: category = Extra, value = 123 *** asyncIterator next *** asyncIterator next *** call find: a1 *** call find: item-5e3f81ed-774a-4f7f-afc5-000a2db34859
client_subscribe.js の出力結果
> node client_subscribe.js *** received event = {"data":{"created":{"category":"Extra","id":"item-5e3f81ed-774a-4f7f-afc5-000a2db34859"}}}
Go言語と Rust で Mutex による排他制御
以下の 3通りを Go 言語と Rust でそれぞれ実装してみました。
サンプルコードは http://github.com/fits/try_samples/tree/master/blog/20201122/
Go 言語の場合
まずは Go 言語による実装です。 Go 言語では goroutine で軽量スレッドによる並行処理を実施できます。
ここでは、goroutine の終了を待機するために sync.WaitGroup を使いました。
WaitGroup では Add した数に相当する Done が呼び出されるまで Wait でブロックして待機する事が可能です。
go_mutex_sample.go
package main import ( "fmt" "sync" ) type Data struct { value int } // (a) func noLock() { var wg sync.WaitGroup var ds []Data for i := 0; i < 100; i++ { wg.Add(1) go func() { ds = append(ds, Data{i}) wg.Done() }() } // goroutine の終了を待機 wg.Wait() fmt.Println("(a) noLock length =", len(ds)) } // (b) func useMutex() { var wg sync.WaitGroup var mu sync.Mutex var ds []Data for i := 0; i < 100; i++ { wg.Add(1) go func() { mu.Lock() ds = append(ds, Data{i}) mu.Unlock() wg.Done() }() } wg.Wait() fmt.Println("(b) useMutex length =", len(ds)) } // (c) func useRWMutex() { var wg sync.WaitGroup var mu sync.RWMutex var ds []Data for i := 0; i < 100; i++ { wg.Add(1) go func() { mu.Lock() ds = append(ds, Data{i}) mu.Unlock() wg.Done() }() } for i := 0; i < 5; i++ { wg.Add(1) go func() { mu.RLock() fmt.Println("(c) progress length =", len(ds)) mu.RUnlock() wg.Done() }() } wg.Wait() fmt.Println("(c) useRWMutex length =", len(ds)) } func main() { noLock() println("-----") useMutex() println("-----") useRWMutex() }
実行結果は以下の通りです。
排他制御を行っていない (a) では、同じ状態の ds に対して複数の goroutine が ds = append(ds, Data{i}) を実行してしまうケースが発生するため、基本的に結果が 100 にはなりません。
実行結果
> go build go_mutex_sample.go > go_mutex_sample (a) noLock length = 95 ----- (b) useMutex length = 100 ----- (c) progress length = 71 (c) progress length = 72 (c) progress length = 72 (c) progress length = 72 (c) progress length = 72 (c) useRWMutex length = 100
Rust の場合
次は Rust による実装です。
ここでは thread::spawn で並行処理を実施し join で待機しています。
基本的に Rust では、Go 言語で実装した noLock のようなスレッドセーフでは無い処理はコンパイルエラーとなって実装できないように工夫されています。
スレッド間で安全に所有権を共有するには、スレッドセーフな参照カウントのポインタである Arc を使用する事になります。
排他制御なしの (a) の処理（下記コードの no_lock）をコンパイルが通るように一応は実装してみましたが、この Arc::get_mut(&mut ds) は常に None を返すので ds.push(Data(i)) の処理は実行されません。
rust_mutex_sample.rs
use std::thread; use std::sync::{Arc, Mutex, RwLock}; struct Data(i32); // (a) fn no_lock() { let mut hs = Vec::new(); let ds = Arc::new(Vec::new()); for i in 0..100 { let mut ds = ds.clone(); hs.push( thread::spawn(move || { if let Some(ds) = Arc::get_mut(&mut ds) { // 上記は常に None となるので下記処理は実行されない ds.push(Data(i)) } }) ); } for h in hs { let _ = h.join(); } println!("(a) no_lock length = {}", ds.len()); } // (b) fn use_mutex() { let mut hs = Vec::new(); let ds = Arc::new(Mutex::new(Vec::new())); for i in 0..100 { let ds = ds.clone(); hs.push( thread::spawn(move || { if let Ok(mut ds) = ds.lock() { ds.push(Data(i)); } }) ); } for h in hs { let _ = h.join(); } println!("(b) use_mutex length = {}", ds.lock().unwrap().len()); } // (c) fn use_rwlock() { let mut hs = Vec::new(); let ds = Arc::new(RwLock::new(Vec::new())); for i in 0..100 { let ds = ds.clone(); hs.push( thread::spawn(move || { if let Ok(mut ds) = ds.write() { ds.push(Data(i)); } }) ); } for _ in 0..5 { let ds = ds.clone(); hs.push( thread::spawn(move || { if let Ok(ds) = ds.read() { println!("(c) progress length = {}", ds.len()); } }) ); } for h in hs { let _ = h.join(); } println!("(c) use_rwlock length = {}", ds.read().unwrap().len()); } fn main() { no_lock(); println!("-----"); use_mutex(); println!("-----"); use_rwlock(); }
実行結果は以下の通りです。
no_lock 内の ds.push(Data(i)) は実行されないので結果は 0 となります。
実行結果
> rustc rust_mutex_sample.rs > rust_mutex_sample (a) no_lock length = 0 ----- (b) use_mutex length = 100 ----- (c) progress length = 99 (c) progress length = 99 (c) progress length = 99 (c) progress length = 99 (c) progress length = 99 (c) use_rwlock length = 100
Node.js で gRPC を試す
「gRPC Server Reflection のクライアント処理」では Node.js で gRPC クライアントを実装しましたが、今回はサーバー側も実装してみます。
サンプルコードは http://github.com/fits/try_samples/tree/master/blog/20201115/
はじめに
gRPC on Node.js では、以下の 2通りの手法が用意されており、それぞれ使用するパッケージが異なります。
- (a) 動的コード生成 （
@grpc/proto-loaderパッケージを使用） - (b) 静的コード生成 （
grpc-toolsパッケージを使用）
更に、gRPC の実装ライブラリとして以下の 2種類が用意されており、どちらかを使う事になります。
- C-based Client and Server （
grpcパッケージを使用） - Pure JavaScript Client （
@grpc/grpc-jsパッケージを使用）
@grpc/grpc-js は現時点で Pure JavaScript Client と表現されていますが、クライアントだけではなくサーバーの実装にも使えます。
ここでは、Pure JavaScript 実装の @grpc/grpc-js を使って、(a) と (b) の両方を試してみます。
サービス定義（proto ファイル）
gRPC のサービス定義として下記ファイルを使用します。
Unary RPC（1リクエスト / 1レスポンス）と Server streaming RPC（1リクエスト / 多レスポンス）、message の oneof、google.protobuf.Empty の扱い等を確認するような内容にしてみました。
proto/item.proto
syntax = "proto3"; import "google/protobuf/empty.proto"; package item; message AddItemRequest { string item_id = 1; uint64 price = 2; } message ItemRequest { string item_id = 1; } message Item { string item_id = 1; uint64 price = 2; } message ItemSubscribeRequest { } message AddedItem { string item_id = 1; uint64 price = 2; } message RemovedItem { string item_id = 1; } message ItemEvent { oneof event { AddedItem added = 1; RemovedItem removed = 2; } } service ItemManage { rpc AddItem(AddItemRequest) returns (google.protobuf.Empty); rpc RemoveItem(ItemRequest) returns (google.protobuf.Empty); rpc GetItem(ItemRequest) returns (Item); rpc Subscribe(ItemSubscribeRequest) returns (stream ItemEvent); }
(a) 動的コード生成（@grpc/proto-loader）
まずは、@grpc/proto-loader を使った動的コード生成を試します。
インストール
@grpc/proto-loader と @grpc/grpc-js をインストールしておきます。
> npm install --save @grpc/proto-loader @grpc/grpc-js
サーバー実装
proto-loader の loadSync 関数 ※ で proto ファイルをロードした結果を grpc-js の loadPackageDefinition で処理する事で型定義などを動的に生成します。
※ 非同期版の load 関数も用意されています
addService で gRPC のサービス定義と処理をマッピングし、bindAsync 後に start を呼び出す事でサーバー処理を開始します。
proto ファイルで定義したメッセージ型と同じフィールドを持つ JavaScript オブジェクトを gRPC のリクエストやレスポンスで使う事ができるようです。
Unary RPC の場合は、第二引数の callback へ失敗時の値と成功時の値をそれぞれ渡す事で処理結果を返します。
任意のエラーを返したい場合は、code で gRPC のステータスコードを、details でエラー内容を指定します。
google.protobuf.Empty の箇所は null もしくは undefined で代用できるようです。
Server streaming RPC の場合は、第一引数（下記コードでは call）の write を呼び出す事で処理結果を返す事ができます。
クライアントが途中で切断したりすると cancelled が発生するようになっており、cancelled 発生後に write を呼び出してもエラー等は発生しないようになっていました。
server.js
const protoLoader = require('@grpc/proto-loader') const grpc = require('@grpc/grpc-js') const protoFile = './proto/item.proto' // proto ファイルのロード const pd = protoLoader.loadSync(protoFile) // gPRC 用の動的な型定義生成 const proto = grpc.loadPackageDefinition(pd) let store = [] let subscribeList = [] const findItem = itemId => store.find(i => i.itemId == itemId) const addItem = (itemId, price) => { if (findItem(itemId)) { return undefined } const item = { itemId, price } store.push(item) return item } const removeItem = itemId => { const item = findItem(itemId) if (item) { store = store.filter(i => i.itemId != item.itemId) } return item } // ItemEvent の配信 const publishEvent = event => { console.log(`*** publish event: ${JSON.stringify(event)}`) subscribeList.forEach(s => s.write(event)) } const server = new grpc.Server() // サービス定義と処理のマッピング server.addService(proto.item.ItemManage.service, { AddItem(call, callback) { const itemId = call.request.itemId const price = call.request.price const item = addItem(itemId, price) if (item) { callback() publishEvent({ added: { itemId, price }}) } else { const err = { code: grpc.status.ALREADY_EXISTS, details: 'exists item' } callback(err) } }, RemoveItem(call, callback) { const itemId = call.request.itemId if (removeItem(itemId)) { callback() publishEvent({ removed: { itemId }}) } else { const err = { code: grpc.status.NOT_FOUND, details: 'item not found' } callback(err) } }, GetItem(call, callback) { const itemId = call.request.itemId const item = findItem(itemId) if (item) { callback(null, item) } else { const err = { code: grpc.status.NOT_FOUND, details: 'item not found' } callback(err) } }, Subscribe(call) { console.log('*** subscribed') subscribeList.push(call) // クライアント切断時の処理 call.on('cancelled', () => { console.log('*** unsubscribed') subscribeList = subscribeList.filter(s => s != call) }) } }) server.bindAsync( '127.0.0.1:50051', grpc.ServerCredentials.createInsecure(), (err, port) => { if (err) { console.error(err) return } console.log(`start server: ${port}`) // 開始 server.start() } )
クライアント実装1
まずは、Unary RPC の API のみ（Subscribe 以外）を呼び出すクライアントを実装してみます。
loadPackageDefinition を実施するところまではサーバーと同じです。
Unary RPC はコールバック関数を伴ったメソッドとして用意されますが、このメソッドに Node.js の util.promisify を直接適用すると不都合が生じたため、Promise 化は自前の関数（下記の promisify）で実施するようにしました。
client.js
const protoLoader = require('@grpc/proto-loader') const grpc = require('@grpc/grpc-js') const protoFile = './proto/item.proto' const pd = protoLoader.loadSync(protoFile) const proto = grpc.loadPackageDefinition(pd) const id = process.argv[2] const client = new proto.item.ItemManage( '127.0.0.1:50051', grpc.credentials.createInsecure() ) // Unary RPC の Promise 化 const promisify = (obj, methodName) => args => new Promise((resolve, reject) => { obj[methodName](args, (err, res) => { if (err) { reject(err) } else { resolve(res) } }) }) const addItem = promisify(client, 'AddItem') const removeItem = promisify(client, 'RemoveItem') const getItem = promisify(client, 'GetItem') const printItem = item => { console.log(`id = ${item.itemId}, price = ${item.price}`) } const run = async () => { await addItem({ itemId: `${id}_item-1`, price: 100 }) const item1 = await getItem({ itemId: `${id}_item-1` }) printItem(item1) await addItem({ itemId: `${id}_item-2`, price: 20 }) const item2 = await getItem({ itemId: `${id}_item-2` }) printItem(item2) await addItem({ itemId: `${id}_item-1`, price: 50 }) .catch(err => console.error(`*** ERROR = ${err.message}`)) await removeItem({ itemId: `${id}_item-1` }) await getItem({ itemId: `${id}_item-1` }) .catch(err => console.error(`*** ERROR = ${err.message}`)) await removeItem({ itemId: `${id}_item-2` }) } run().catch(err => console.error(err))
クライアント実装2
次は Server streaming RPC のクライアント実装です。
client_subscribe.js
const protoLoader = require('@grpc/proto-loader') const grpc = require('@grpc/grpc-js') const protoFile = './proto/item.proto' const pd = protoLoader.loadSync(protoFile) const proto = grpc.loadPackageDefinition(pd) const client = new proto.item.ItemManage( '127.0.0.1:50051', grpc.credentials.createInsecure() ) const stream = client.Subscribe({}) // メッセージ受信時 stream.on('data', event => { console.log(`*** received event = ${JSON.stringify(event)}`) }) // サーバー終了時 stream.on('end', () => console.log('*** stream end')) stream.on('error', err => console.log(`*** stream error: ${err}`))
動作確認
server.js の実行後、client_subscribe.js を 2つ起動した後に client.js を実行してみます。
Server 実行
> node server.js start server: 50051
Client2-1 実行
> node client_subscribe.js
Client2-2 実行
> node client_subscribe.js
Client1 実行
> node client.js a1 id = a1_item-1, price = 100 id = a1_item-2, price = 20 *** ERROR = 6 ALREADY_EXISTS: exists item *** ERROR = 5 NOT_FOUND: item not found
この時点で出力内容は以下のようになりました。
Server 出力内容
> node server.js
start server: 50051
*** subscribed
*** subscribed
*** publish event: {"added":{"itemId":"a1_item-1","price":100}}
*** publish event: {"added":{"itemId":"a1_item-2","price":20}}
*** publish event: {"removed":{"itemId":"a1_item-1"}}
*** publish event: {"removed":{"itemId":"a1_item-2"}}
Client2-1、Client2-2 出力内容
> node client_subscribe.js
*** received event = {"added":{"itemId":"a1_item-1","price":{"low":100,"high":0,"unsigned":true}}}
*** received event = {"added":{"itemId":"a1_item-2","price":{"low":20,"high":0,"unsigned":true}}}
*** received event = {"removed":{"itemId":"a1_item-1"}}
*** received event = {"removed":{"itemId":"a1_item-2"}}
Client2-2 を Ctrl + c で終了後に、Server も Ctrl + c で終了すると以下のようになり、Client2-1 のプロセスは終了しました。
Server 出力内容
> node server.js
start server: 50051
・・・
*** publish event: {"removed":{"itemId":"a1_item-2"}}
*** unsubscribed
^C
Client2-1 出力内容
> node client_subscribe.js
・・・
*** received event = {"removed":{"itemId":"a1_item-2"}}
*** stream error: Error: 13 INTERNAL: Received RST_STREAM with code 2 (Internal server error)
*** stream end
特に問題はなく、正常に動作しているようです。
(b) 静的コード生成（grpc-tools）
grpc-tools を使った静的コード生成を試します。
インストールとコード生成
grpc-tools、@grpc/grpc-js、google-protobuf をインストールしておきます。
> npm install --save-dev grpc-tools ・・・ > npm install --save @grpc/grpc-js google-protobuf ・・・
grpc-tools をインストールする事で使えるようになる grpc_tools_node_protoc コマンドで proto ファイルからコードを生成します。
grpc_tools_node_protoc コマンドは内部的に protoc コマンドを grpc_node_plugin プラグインを伴って呼び出すようになっています。
--grpc_out でサービス定義用のファイル xxx_grpc_pb.js が生成され、--js_out でメッセージ定義用のファイルが生成されます。
サービス定義 xxx_grpc_pb.js は --js_out で import_style=commonjs オプションを指定する事を前提としたコードになっています。※
※ import_style=commonjs オプションを指定した際に生成される
xxx_pb.js を参照するようになっている
また、--grpc_out はデフォルトで grpc パッケージ用のコードを生成するため、ここでは grpc_js オプションを指定して @grpc/grpc-js 用のコードを生成するようにしています。
静的コード生成例（grpc_tools_node_protoc コマンド）
> mkdir generated > grpc_tools_node_protoc --grpc_out=grpc_js:generated --js_out=import_style=commonjs:generated proto/*.proto ・・・
サーバー実装
(a) の場合と処理内容に大きな違いはありませんが、リクエストやレスポンスでは生成された型を使います。
アクセサメソッド（getter、setter）で値の取得や設定ができるようになっており、new 時に配列として全フィールドの値を指定する事もできるようです。
JavaScript オブジェクトへ変換したい場合は toObject メソッドを使用します。
addService でマッピングする際のメソッド名の一文字目が小文字になっています。
proto ファイルで定義したサービス名の後に Service を付けた型（ここでは ItemManageService）がサーバー処理用、Client を付けた型がクライアント処理用の型定義となるようです。
server.js
const grpc = require('@grpc/grpc-js') const { Item, AddedItem, RemovedItem, ItemEvent } = require('./generated/proto/item_pb') const { ItemManageService } = require('./generated/proto/item_grpc_pb') const { Empty } = require('google-protobuf/google/protobuf/empty_pb') let store = [] let subscribeList = [] const findItem = itemId => store.find(i => i.getItemId() == itemId) const addItem = (itemId, price) => { if (findItem(itemId)) { return undefined } const item = new Item([itemId, price]) store.push(item) return item } const removeItem = itemId => { const item = findItem(itemId) if (item) { store = store.filter(i => i.getItemId() != item.getItemId()) } return item } const createAddedEvent = (itemId, price) => { const event = new ItemEvent() event.setAdded(new AddedItem([itemId, price])) return event } const createRemovedEvent = itemId => { const event = new ItemEvent() event.setRemoved(new RemovedItem([itemId])) return event } const publishEvent = event => { // toObject で JavaScript オブジェクトへ変換 console.log(`*** publish event: ${JSON.stringify(event.toObject())}`) subscribeList.forEach(s => s.write(event)) } const server = new grpc.Server() server.addService(ItemManageService, { addItem(call, callback) { const itemId = call.request.getItemId() const price = call.request.getPrice() const item = addItem(itemId, price) if (item) { callback(null, new Empty()) publishEvent(createAddedEvent(itemId, price)) } else { const err = { code: grpc.status.ALREADY_EXISTS, details: 'exists item' } callback(err) } }, removeItem(call, callback) { const itemId = call.request.getItemId() if (removeItem(itemId)) { callback(null, new Empty()) publishEvent(createRemovedEvent(itemId)) } else { const err = { code: grpc.status.NOT_FOUND, details: 'item not found' } callback(err) } }, getItem(call, callback) { const itemId = call.request.getItemId() const item = findItem(itemId) if (item) { callback(null, item) } else { const err = { code: grpc.status.NOT_FOUND, details: 'item not found' } callback(err) } }, subscribe(call) { console.log('*** subscribed') subscribeList.push(call) call.on('cancelled', () => { console.log('*** unsubscribed') subscribeList = subscribeList.filter(s => s != call) }) } }) server.bindAsync( ・・・ )
クライアント実装1
生成された型を使う点とメソッド名の先頭が小文字になっている点を除くと、基本的に (a) と同じです。
client.js
const grpc = require('@grpc/grpc-js') const { AddItemRequest, ItemRequest } = require('./generated/proto/item_pb') const { ItemManageClient } = require('./generated/proto/item_grpc_pb') const id = process.argv[2] const client = new ItemManageClient( '127.0.0.1:50051', grpc.credentials.createInsecure() ) const promisify = (obj, methodName) => args => new Promise((resolve, reject) => { ・・・ }) const addItem = promisify(client, 'addItem') const removeItem = promisify(client, 'removeItem') const getItem = promisify(client, 'getItem') const printItem = item => { console.log(`id = ${item.getItemId()}, price = ${item.getPrice()}`) } const run = async () => { await addItem(new AddItemRequest([`${id}_item-1`, 100])) const item1 = await getItem(new ItemRequest([`${id}_item-1`])) printItem(item1) await addItem(new AddItemRequest([`${id}_item-2`, 20])) const item2 = await getItem(new ItemRequest([`${id}_item-2`])) printItem(item2) await addItem(new AddItemRequest([`${id}_item-1`, 50])) .catch(err => console.error(`*** ERROR = ${err.message}`)) await removeItem(new ItemRequest([`${id}_item-1`])) await getItem(new ItemRequest([`${id}_item-1`])) .catch(err => console.error(`*** ERROR = ${err.message}`)) await removeItem(new ItemRequest([`${id}_item-2`])) } run().catch(err => console.error(err))
クライアント実装2
こちらも同様です。
client_subscribe.js
const grpc = require('@grpc/grpc-js') const { ItemSubscribeRequest } = require('./generated/proto/item_pb') const { ItemManageClient } = require('./generated/proto/item_grpc_pb') const client = new ItemManageClient( ・・・ ) const stream = client.subscribe(new ItemSubscribeRequest()) stream.on('data', event => { // toObject で JavaScript オブジェクトへ変換 console.log(`*** received event = ${JSON.stringify(event.toObject())}`) }) ・・・
動作確認
(a) と同じ操作を行った結果は以下のようになりました。
Server 出力内容
> node server.js
start server: 50051
*** subscribed
*** subscribed
*** publish event: {"added":{"itemId":"a1_item-1","price":100}}
*** publish event: {"added":{"itemId":"a1_item-2","price":20}}
*** publish event: {"removed":{"itemId":"a1_item-1"}}
*** publish event: {"removed":{"itemId":"a1_item-2"}}
*** unsubscribed
^C
Client1 出力内容
> node client.js a1 id = a1_item-1, price = 100 id = a1_item-2, price = 20 *** ERROR = 6 ALREADY_EXISTS: exists item *** ERROR = 5 NOT_FOUND: item not found
Client2-1 出力内容
> node client_subscribe.js
*** received event = {"added":{"itemId":"a1_item-1","price":100}}
*** received event = {"added":{"itemId":"a1_item-2","price":20}}
*** received event = {"removed":{"itemId":"a1_item-1"}}
*** received event = {"removed":{"itemId":"a1_item-2"}}
*** stream error: Error: 13 INTERNAL: Received RST_STREAM with code 2 (Internal server error)
*** stream end
Client2-2 出力内容
> node client_subscribe.js
*** received event = {"added":{"itemId":"a1_item-1","price":100}}
*** received event = {"added":{"itemId":"a1_item-2","price":20}}
*** received event = {"removed":{"itemId":"a1_item-1"}}
*** received event = {"removed":{"itemId":"a1_item-2"}}
^C
(a) と (b) は同一の gRPC サービス（proto ファイル）を実装したものなので当然ですが、(a) と (b) を相互接続しても特に問題はありませんでした。
RLlib を使ってナップサック問題を強化学習2
局所最適に陥っていたと思われる 前回 に対して、以下の改善案 ※ を思いついたので試してみました。
- より困難な目標を達成した場合に報酬（価値）へボーナスを加算
※ 局所最適から脱して、より良い結果を目指す効果を期待
今回のサンプルコードは http://github.com/fits/try_samples/tree/master/blog/20201019/
サンプル1 改良版（ボーナス加算）
単一操作（品物の 1つを -1 or +1 するか何もしない）を行動とした（前回の）サンプル1 にボーナスを加算する処理を加えてみました。
とりあえず、価値の合計が 375（0-1 ナップサック問題としての最適解）を超えた場合に報酬へ +200 するようにしてみます。
前回から、変数や関数名を一部変更していますが、基本的な処理内容に変更はありません。
また、PPOTrainer では episode_reward_mean / vf_clip_param の値が 200 を超えると警告ログを出すようなので（ppo.py の warn_about_bad_reward_scales）、config で vf_clip_param（デフォルト値は 10）の値を変更するようにしています。
sample1_bonus.ipynb
・・・ def next_state(items, state, action): idx = action // 2 act = action % 2 if idx < len(items): state[idx] += (1 if act == 1 else -1) return state def calc_value(items, state, max_weight, burst_value): reward = 0 weight = 0 for i in range(len(state)): reward += items[i][0] * state[i] weight += items[i][1] * state[i] if weight > max_weight or min(state) < 0: reward = burst_value return reward, weight class Knapsack(gym.Env): def __init__(self, config): self.items = config["items"] self.max_weight = config["max_weight"] self.episode_steps = config["episode_steps"] self.burst_reward = config["burst_reward"] self.bonus_rules = config["bonus_rules"] n = self.episode_steps self.action_space = Discrete(len(self.items) * 2 + 1) self.observation_space = Box(low = -n, high = n, shape = (len(self.items), )) self.reset() def reset(self): self.current_steps = 0 self.state = [0 for _ in self.items] return self.state def step(self, action): self.state = next_state(self.items, self.state, action) r, _ = calc_value(self.items, self.state, self.max_weight, self.burst_reward) reward = r # 段階的なボーナス加算 for (v, b) in self.bonus_rules: if r > v: reward += b self.current_steps += 1 done = self.current_steps >= self.episode_steps return self.state, reward, done, {} items = [ [105, 10], [74, 7], [164, 15], [32, 3], [235, 22] ] config = { "env": Knapsack, "vf_clip_param": 60, "env_config": { "items": items, "episode_steps": 10, "max_weight": 35, "burst_reward": -100, # ボーナスの設定 "bonus_rules": [ (375, 200) ] } } ・・・ trainer = PPOTrainer(config = config) ・・・ # 30回の学習 for _ in range(30): r = trainer.train() ・・・ ・・・ rs = [] # 1000回試行 for _ in range(1000): s = [0 for _ in range(len(items))] r_tmp = config["env_config"]["burst_reward"] for _ in range(config["env_config"]["episode_steps"]): a = trainer.compute_action(s) s = next_state(items, s, a) r, w = calc_value(items, s, config["env_config"]["max_weight"], config["env_config"]["burst_reward"]) r_tmp = max(r, r_tmp) rs.append(r_tmp) collections.Counter(rs)
上記の結果（30回の学習後に 1000回試行してそれぞれの最高値をカウント）は以下のようになりました。
結果
Counter({376: 957, 375: 42, 334: 1})
最適解の 376 が出るようになっており、ボーナスの効果はそれなりにありそうです。
ただし、毎回このような結果になるわけではなく、前回と同じように 375（0-1 ナップサック問題としての最適解）止まりとなる場合もあります。
検証
次に、ナップサック問題の内容を変えて検証してみます。
ここでは、「2.5 ナップサック問題 - 数理システム」の例題を題材として、状態の範囲やボーナスの内容を変えると結果にどのような差が生じるのかを確認します。
ナップサック問題の内容
| 価値 | サイズ |
|---|---|
| 120 | 10 |
| 130 | 12 |
| 80 | 7 |
| 100 | 9 |
| 250 | 21 |
| 185 | 16 |
最大容量（サイズ） 65 における最適解は以下の通りです。
価値 770 の組み合わせ（最適解）
3, 0, 2, 0, 1, 0
価値 745 の組み合わせ（0-1 ナップサック問題の最適解）
0, 1, 1, 1, 1, 1
1. 単一操作（品物の 1つを -1 or +1 するか何もしない）
行動は前回の サンプル1 と同様の以下とします。
- 品物のどれか 1つを -1 or +1 するか、何も変更しない
ここでは、以下のような状態範囲とボーナスを試しました。
状態範囲（品物毎の個数の範囲）
| 状態タイプ | 最小値 | 最大値 |
|---|---|---|
| a | -10 | 10 |
| b | 0 | 5 |
| c | 0 | 3 |
ボーナス定義
| ボーナス定義タイプ | v > 750 | v > 760 | v > 765 |
|---|---|---|---|
| 0 | 0 | 0 | 0 |
| 1 | 100 | 100 | 100 |
| 2 | 100 | 200 | 400 |
ボーナスは段階的に加算し、ボーナス定義タイプ 2 で価値が仮に 770 だった場合は、700（100 + 200 + 400）を加算する事にします。
また、状態（品物毎の個数）はその範囲を超えないよう最小値もしくは最大値で止まるようにしました。
なお、ここからは Jupyter Notebook ではなく Python スクリプトとして実行します。
test1.py
import sys import numpy as np import gym from gym.spaces import Discrete, Box import ray from ray.rllib.agents.ppo import PPOTrainer import collections N = int(sys.argv[1]) EPISODE_STEPS = int(sys.argv[2]) STATE_TYPE = sys.argv[3] BONUS_TYPE = sys.argv[4] items = [ [120, 10], [130, 12], [80, 7], [100, 9], [250, 21], [185, 16] ] state_types = { "a": (-10, 10), "b": (0, 5), "c": (0, 3) } bonus_types = { "0": [], "1": [(750, 100), (760, 100), (765, 100)], "2": [(750, 100), (760, 200), (765, 400)] } vf_clip_params = { "0": 800, "1": 1100, "2": 1500 } def next_state(items, state, action, state_range): idx = action // 2 act = action % 2 if idx < len(items): v = state[idx] + (1 if act == 1 else -1) # 状態が範囲内に収まるように調整 state[idx] = min(state_range[1], max(state_range[0], v)) return state def calc_value(items, state, max_weight, burst_value): reward = 0 weight = 0 for i in range(len(state)): reward += items[i][0] * state[i] weight += items[i][1] * state[i] if weight > max_weight or min(state) < 0: reward = burst_value return reward, weight class Knapsack(gym.Env): def __init__(self, config): self.items = config["items"] self.max_weight = config["max_weight"] self.episode_steps = config["episode_steps"] self.burst_reward = config["burst_reward"] self.state_range = config["state_range"] self.bonus_rules = config["bonus_rules"] self.action_space = Discrete(len(self.items) * 2 + 1) self.observation_space = Box( low = self.state_range[0], high = self.state_range[1], shape = (len(self.items), ) ) self.reset() def reset(self): self.current_steps = 0 self.state = [0 for _ in self.items] return self.state def step(self, action): self.state = next_state(self.items, self.state, action, self.state_range) r, _ = calc_value(self.items, self.state, self.max_weight, self.burst_reward) reward = r for (v, b) in self.bonus_rules: if r > v: reward += b self.current_steps += 1 done = self.current_steps >= self.episode_steps return self.state, reward, done, {} config = { "env": Knapsack, "vf_clip_param": vf_clip_params[BONUS_TYPE], "env_config": { "items": items, "max_weight": 65, "burst_reward": -100, "episode_steps": EPISODE_STEPS, "state_range": state_types[STATE_TYPE], "bonus_rules": bonus_types[BONUS_TYPE] } } ray.init() trainer = PPOTrainer(config = config) for _ in range(N): r = trainer.train() print(f'iter = {r["training_iteration"]}') print(f'N = {N}, EPISODE_STEPS = {EPISODE_STEPS}, state_type = {STATE_TYPE}, bonus_type = {BONUS_TYPE}') rs = [] for _ in range(1000): s = [0 for _ in range(len(items))] r_tmp = config["env_config"]["burst_reward"] for _ in range(config["env_config"]["episode_steps"]): a = trainer.compute_action(s) s = next_state(items, s, a, config["env_config"]["state_range"]) r, w = calc_value( items, s, config["env_config"]["max_weight"], config["env_config"]["burst_reward"] ) r_tmp = max(r, r_tmp) rs.append(r_tmp) print( collections.Counter(rs) ) ray.shutdown()
実行例
> python test1.py 50 10 a 0
学習回数 50、1エピソードのステップ数 10 で学習した後、1000回の試行で最も件数の多かった価値を列挙する処理を 3回実施した結果です。（() 内の数値は 1000回の内にその値が最高値だった件数）
結果（学習回数 = 50、エピソードのステップ数 = 10）
| 状態タイプ | ボーナス定義タイプ | 状態の最小値 | 状態の最大値 | 765 超過時の総ボーナス | 1回目 | 2回目 | 3回目 | |
|---|---|---|---|---|---|---|---|---|
| a-0 | a | 0 | -10 | 10 | 0 | 735 (994) | 735 (935) | 735 (916) |
| a-1 | a | 1 | -10 | 10 | +300 | 745 (965) | 745 (977) | 735 (976) |
| a-2 | a | 2 | -10 | 10 | +700 | 735 (945) | 735 (971) | 770 (1000) |
| b-0 | b | 0 | 0 | 5 | 0 | 750 (931) | 750 (829) | 750 (1000) |
| b-1 | b | 1 | 0 | 5 | +300 | 765 (995) | 765 (998) | 750 (609) |
| b-2 | b | 2 | 0 | 5 | +700 | 765 (1000) | 765 (995) | 765 (998) |
| c-0 | c | 0 | 0 | 3 | 0 | 750 (998) | 750 (996) | 750 (1000) |
| c-1 | c | 1 | 0 | 3 | +300 | 765 (1000) | 750 (993) | 765 (1000) |
| c-2 | c | 2 | 0 | 3 | +700 | 765 (999) | 765 (1000) | 770 (999) |
やはり、ボーナスは有効そうですが、状態タイプ a のように状態の範囲が広く、ボーナスの発生頻度が低くなるようなケースでは有効に働かない可能性も高そうです。
ボーナス定義タイプ 2 で最適解の 770 が出るようになっているものの、頻出するようなものでも無く、たまたま学習が上手くいった場合にのみ発生しているような印象でした。
また、b-1 の 3回目で件数が他と比べて低くなっていますが、こちらは学習が（順調に進まずに）足りていない状態だと考えられます。
2. 一括操作（全品物をそれぞれ -1 or 0 or +1 する）
次に、行動を以下のように変えて同じように検証してみます。
- 全ての品物を対象にそれぞれを -1 or 0 or +1 する
こちらは、ボーナス加算タイプを 1種類追加しました。
状態範囲（品物毎の個数の範囲）
| 状態タイプ | 最小値 | 最大値 |
|---|---|---|
| a | -10 | 10 |
| b | 0 | 5 |
| c | 0 | 3 |
ボーナス加算
| ボーナス加算タイプ | v > 750 | v > 760 | v > 765 |
|---|---|---|---|
| 0 | 0 | 0 | 0 |
| 1 | 100 | 100 | 100 |
| 2 | 100 | 200 | 400 |
| 3 | 200 | 400 | 800 |
行動の変更に伴い action_space を Box で定義しています。
test2.py
・・・
state_types = {
"a": (-10, 10),
"b": (0, 5),
"c": (0, 3)
}
bonus_types = {
"0": [],
"1": [(750, 100), (760, 100), (765, 100)],
"2": [(750, 100), (760, 200), (765, 400)],
"3": [(750, 200), (760, 400), (765, 800)]
}
・・・
def next_state(items, state, action, state_range):
for i in range(len(action)):
v = state[i] + round(action[i])
state[i] = min(state_range[1], max(state_range[0], v))
return state
・・・
class Knapsack(gym.Env):
def __init__(self, config):
self.items = config["items"]
self.max_weight = config["max_weight"]
self.episode_steps = config["episode_steps"]
self.burst_reward = config["burst_reward"]
self.state_range = config["state_range"]
self.bonus_rules = config["bonus_rules"]
# 品物毎の -1 ～ 1
self.action_space = Box(low = -1, high = 1, shape = (len(self.items), ))
self.observation_space = Box(
low = self.state_range[0],
high = self.state_range[1],
shape = (len(self.items), )
)
self.reset()
def reset(self):
self.current_steps = 0
self.state = [0 for _ in self.items]
return self.state
def step(self, action):
self.state = next_state(self.items, self.state, action, self.state_range)
r, _ = calc_value(self.items, self.state, self.max_weight, self.burst_reward)
reward = r
for (v, b) in self.bonus_rules:
if r > v:
reward += b
self.current_steps += 1
done = self.current_steps >= self.episode_steps
return self.state, reward, done, {}
・・・
こちらの方法では、学習回数 50 では明らかに足りなかったので 100 にして実施しました。
結果（学習回数 = 100、エピソードのステップ数 = 10）
| 状態タイプ | ボーナス加算タイプ | 状態の最小値 | 状態の最大値 | 765 超過時の総ボーナス | 1回目 | 2回目 | 3回目 | |
|---|---|---|---|---|---|---|---|---|
| a-0 | a | 0 | -10 | 10 | 0 | 735 (477) | 735 (531) | 735 (714) |
| a-1 | a | 1 | -10 | 10 | +300 | 735 (689) | 735 (951) | 745 (666) |
| a-2 | a | 2 | -10 | 10 | +700 | 735 (544) | 735 (666) | 735 (719) |
| a-3 | a | 3 | -10 | 10 | +1400 | 745 (633) | 735 (735) | 735 (875) |
| b-0 | b | 0 | 0 | 5 | 0 | 735 (364) | 760 (716) | 740 (590) |
| b-1 | b | 1 | 0 | 5 | +300 | 735 (935) | 760 (988) | 655 (685) |
| b-2 | b | 2 | 0 | 5 | +700 | 760 (1000) | 735 (310) | 770 (963) |
| b-3 | b | 3 | 0 | 5 | +1400 | 675 (254) | 770 (1000) |
770 (909) |
| c-0 | c | 0 | 0 | 3 | 0 | 735 (762) | 740 (975) | 740 (669) |
| c-1 | c | 1 | 0 | 3 | +300 | 740 (935) | 740 (842) | 735 (963) |
| c-2 | c | 2 | 0 | 3 | +700 | 770 (999) |
770 (1000) |
715 (508) |
| c-3 | c | 3 | 0 | 3 | +1400 | 770 (1000) |
770 (1000) |
770 (1000) |
学習の足りていない所が散見されますが（学習も不安定）、特定のタイプで最適解の 770 が割と頻繁に出るようになりました。
ただ、c-3 の場合でも 770 が出やすくなっているものの、確実にそのように学習するわけではありませんでした。
結局のところ、状態・行動・報酬の設計次第という事かもしれません。
RLlib を使ってナップサック問題を強化学習
ナップサック問題へ強化学習を適用すると、どうなるのか気になったので試してみました。
強化学習には、Ray に含まれている RLlib を使い、Jupyter Notebook 上で実行します。
今回のサンプルコードは http://github.com/fits/try_samples/tree/master/blog/20200922/
はじめに
以下のようにして Ray と RLlib をインストールしておきます。（TensorFlow も事前にインストールしておく）
Ray インストール
> pip install ray[rllib]
ナップサック問題
今回は、以下のような価値と重さを持った品物に対して、重さの合計が 35 以下で価値の合計を最大化する品物の組み合わせを探索する事にします。
| 価値 | 重さ |
|---|---|
| 105 | 10 |
| 74 | 7 |
| 164 | 15 |
| 32 | 3 |
| 235 | 22 |
品物をそれぞれ 1個までしか選べない場合（0-1 ナップサック問題）の最適な組み合わせは、以下のように 5番目以外を 1個ずつ選ぶ事です。（価値の合計は 375）
価値 375 の組み合わせ（0-1 ナップサック問題の最適解）
1, 1, 1, 1, 0
また、同じ品物をいくらでも選べる場合の最適な組み合わせは以下のようになります。（価値の合計は 376）
価値 376 の組み合わせ
0, 2, 1, 2, 0
強化学習でこのような組み合わせを導き出す事ができるのか確認します。
1. サンプル1 - sample1.ipynb
とりあえず、強化学習における状態・行動・報酬を以下のようにしてみました。 エピソードは、指定した回数（今回は 10回）の行動を行う事で終了とします。
| 状態 | 行動 | （即時）報酬 |
|---|---|---|
| 品物毎の個数 | 品物の個数を操作（-1, +1） | 価値の合計 |
行動は以下のような 0 ～ 10 の数値で表現する事にします。
- 0 = 1番目の品物の個数を -1
- 1 = 1番目の品物の個数を +1
- ・・・
- 8 = 5番目の品物の個数を -1
- 9 = 5番目の品物の個数を +1
- 10 = 個数を変更しない（現状維持）
これらを OpenAI Gym で定義したのが次のコードです。
環境は gym.Env を継承し、__init__ で行動空間（action_space）と状態空間（observation_space）を定義、reset で状態の初期化、step で状態の更新と報酬の算出を行うように実装します。
step の戻り値は、更新後の状態、報酬、エピソード終了か否か、（デバッグ用途等の）情報 となっています。
Discrete(n) は 0 ～ n - 1 の整数値、Box は low ～ high の実数値の多次元配列となっており、、行動空間と状態空間の定義にそれぞれ使用しています。
環境定義
import gym from gym.spaces import Discrete, Box def next_state(items, state, action): idx = action // 2 act = action % 2 if idx < len(items): state[idx] += (1 if act == 1 else -1) return state # 報酬の算出 def calc_reward(items, state, max_weight, burst_reward): reward = 0 weight = 0 for i in range(len(state)): reward += items[i][0] * state[i] weight += items[i][1] * state[i] if weight > max_weight or min(state) < 0: reward = burst_reward return reward, weight class Knapsack(gym.Env): def __init__(self, config): self.items = config["items"] # 重さの上限値 self.max_weight = config["max_weight"] # 行動の回数 self.max_count = config["max_count"] # 重さが超過するか、個数が負の数となった場合の報酬 self.burst_reward = config["burst_reward"] n = self.max_count # 行動空間の定義 self.action_space = Discrete(len(self.items) * 2 + 1) # 状態空間の定義 self.observation_space = Box(low = -n, high = n, shape = (len(self.items), )) self.reset() def reset(self): self.count = 0 self.state = [0 for _ in self.items] return self.state def step(self, action): # 状態の更新 self.state = next_state(self.items, self.state, action) # 報酬の算出 reward, _ = calc_reward(self.items, self.state, self.max_weight, self.burst_reward) self.count += 1 # エピソード完了の判定 done = self.count >= self.max_count return self.state, reward, done, {}
次に上記環境のための設定を行います。
env_config の内容が __init__ の config 引数となります。
重さの上限値を超えた場合や個数が負の数となった場合の報酬（burst_reward）をとりあえず -100 としています。
なお、基本的に RLlib のデフォルト設定値を使う事にします。
設定
items = [
[105, 10],
[74, 7],
[164, 15],
[32, 3],
[235, 22]
]
config = {
"env": Knapsack,
"env_config": {"items": items, "max_count": 10, "max_weight": 35, "burst_reward": -100}
}
強化学習を実施する前に、Ray を初期化しておきます。
Ray 初期化
import ray
ray.init()
(a) PPO（Proximal Policy Optimization）
PPO アルゴリズムを試してみます。
トレーナーの定義 - PPO
from ray.rllib.agents.ppo import PPOTrainer trainer = PPOTrainer(config = config)
まずは、学習（train の実行）を 10回繰り返してみます。
後で学習時の状況を確認するために episode_reward_max 等の値を保持するようにしています。
学習
r_max = [] r_min = [] r_mean = []
from ray.tune.logger import pretty_print for _ in range(10): r = trainer.train() print(pretty_print(r)) r_max.append(r["episode_reward_max"]) # 最大 r_min.append(r["episode_reward_min"]) # 最小 r_mean.append(r["episode_reward_mean"]) # 平均
以下のコードで結果を確認してみます。
compute_action を呼び出す事で、指定した状態に対する行動を取得できます。
評価1
s = [0 for _ in range(len(items))] for _ in range(config["env_config"]["max_count"]): a = trainer.compute_action(s) s = next_state(items, s, a) r, w = calc_reward(items, s, config["env_config"]["max_weight"], config["env_config"]["burst_reward"]) print(f"{a}, {s}, {r}, {w}")
下記のように、価値の合計が 375 となる組み合わせ（0-1 ナップサック問題とした場合の最適解）が現れており、ある程度は学習できているように見えます。
評価1の結果 - PPO 学習 10回
1, [1, 0, 0, 0, 0], 105, 10 3, [1, 1, 0, 0, 0], 179, 17 5, [1, 1, 1, 0, 0], 343, 32 7, [1, 1, 1, 1, 0], 375, 35 7, [1, 1, 1, 2, 0], -100, 38 6, [1, 1, 1, 1, 0], 375, 35 0, [0, 1, 1, 1, 0], 270, 25 1, [1, 1, 1, 1, 0], 375, 35 6, [1, 1, 1, 0, 0], 343, 32 7, [1, 1, 1, 1, 0], 375, 35
これだけだとよく分からないので、100回繰り返してそれぞれの報酬の最高値をカウントしてみます。
評価2
import collections rs = [] for _ in range(100): s = [0 for _ in range(len(items))] r_tmp = config["env_config"]["burst_reward"] for _ in range(config["env_config"]["max_count"]): a = trainer.compute_action(s) s = next_state(items, s, a) r, w = calc_reward(items, s, config["env_config"]["max_weight"], config["env_config"]["burst_reward"]) r_tmp = max(r, r_tmp) rs.append(r_tmp) collections.Counter(rs)
結果は以下のようになりました。 最高値の 376 ではなく 375 の組み合わせへ向かうように学習が進んでいるように見えます。
評価2の結果 - PPO 学習 10回
Counter({302: 6,
309: 2,
284: 1,
343: 12,
375: 39,
270: 4,
341: 7,
373: 1,
340: 2,
269: 1,
372: 6,
301: 4,
376: 1,
334: 2,
333: 2,
344: 1,
299: 4,
317: 1,
275: 1,
349: 1,
316: 1,
312: 1})
更に、学習を 10回繰り返した後の結果です。 375 へ向かって収束しているように見えます。
評価1の結果 - PPO 学習 20回
1, [1, 0, 0, 0, 0], 105, 10 5, [1, 0, 1, 0, 0], 269, 25 3, [1, 1, 1, 0, 0], 343, 32 7, [1, 1, 1, 1, 0], 375, 35 10, [1, 1, 1, 1, 0], 375, 35 10, [1, 1, 1, 1, 0], 375, 35 10, [1, 1, 1, 1, 0], 375, 35 2, [1, 0, 1, 1, 0], 301, 28 3, [1, 1, 1, 1, 0], 375, 35 0, [0, 1, 1, 1, 0], 270, 25
評価2の結果 - PPO 学習 20回
Counter({375: 69, 373: 18, 376: 3, 341: 5, 372: 1, 302: 1, 344: 3})
50回学習した後の結果は以下のようになりました。
評価2の結果 - PPO 学習 50回
Counter({375: 98, 373: 2})
ここまでの（学習時の）報酬状況をグラフ化してみます。
学習回数と報酬のグラフ化
%matplotlib inline import matplotlib.pyplot as plt plt.plot(r_max, label = "reward_max", color = "red") plt.plot(r_min, label = "reward_min", color = "green") plt.plot(r_mean, label = "reward_mean", color = "blue") plt.legend(loc = "upper left") plt.ylim([-1000, 3700]) plt.ylabel("reward") plt.show()
学習時の報酬グラフ - PPO 学習 50回
エピソード内の報酬を高めていくには重量超過やマイナス個数を避けるのが重要、それには各品物の個数を 0 か 1 にしておくのが無難なため、0-1 ナップサック問題としての最適解へ向かっていくのかもしれません。
(b) DQN（Deep Q-Network）
比較のために DQN でも実行してみます。
PPOTrainer の代わりに DQNTrainer を使うだけです。
トレーナーの定義 - DQN
from ray.rllib.agents.dqn import DQNTrainer trainer = DQNTrainer(config = config)
10回の学習では以下のような結果となりました。
評価1の結果 - DQN 学習 10回
5, [0, 0, 1, 0, 0], 164, 15 7, [0, 0, 1, 1, 0], 196, 18 2, [0, -1, 1, 1, 0], -100, 11 5, [0, -1, 2, 1, 0], -100, 26 2, [0, -2, 2, 1, 0], -100, 19 0, [-1, -2, 2, 1, 0], -100, 9 1, [0, -2, 2, 1, 0], -100, 19 3, [0, -1, 2, 1, 0], -100, 26 5, [0, -1, 3, 1, 0], -100, 41 6, [0, -1, 3, 0, 0], -100, 38
評価2の結果 - DQN 学習 10回
Counter({-100: 33,
360: 1,
105: 8,
270: 2,
372: 5,
315: 1,
0: 5,
374: 1,
74: 2,
274: 1,
235: 7,
340: 3,
32: 4,
164: 5,
238: 1,
309: 2,
106: 1,
228: 2,
169: 1,
267: 3,
343: 1,
196: 1,
331: 1,
363: 1,
254: 1,
299: 1,
317: 1,
212: 1,
301: 1,
328: 1,
138: 1,
64: 1})
DQN は PPO に比べて episodes_total（学習で実施したエピソード数の合計）が 1/4 程度と少なかったので、40回まで実施してみました。
評価1の結果 - DQN 学習 40回
6, [0, 0, 0, -1, 0], -100, -3 7, [0, 0, 0, 0, 0], 0, 0 1, [1, 0, 0, 0, 0], 105, 10 4, [1, 0, -1, 0, 0], -100, -5 4, [1, 0, -2, 0, 0], -100, -20 8, [1, 0, -2, 0, -1], -100, -42 7, [1, 0, -2, 1, -1], -100, -39 1, [2, 0, -2, 1, -1], -100, -29 2, [2, -1, -2, 1, -1], -100, -36 4, [2, -1, -3, 1, -1], -100, -51
評価2の結果 - DQN 学習 40回
Counter({0: 5,
328: 4,
164: 8,
-100: 28,
340: 5,
228: 1,
309: 4,
74: 1,
238: 4,
235: 7,
32: 3,
358: 2,
343: 2,
196: 2,
269: 3,
372: 2,
365: 1,
179: 2,
374: 1,
302: 1,
106: 3,
312: 1,
105: 5,
270: 2,
148: 2,
360: 1})
80回学習した結果です。
評価1 - DQN 学習 80回
5, [0, 0, 1, 0, 0], 164, 15 1, [1, 0, 1, 0, 0], 269, 25 7, [1, 0, 1, 1, 0], 301, 28 4, [1, 0, 0, 1, 0], 137, 13 0, [0, 0, 0, 1, 0], 32, 3 2, [0, -1, 0, 1, 0], -100, -4 7, [0, -1, 0, 2, 0], -100, -1 1, [1, -1, 0, 2, 0], -100, 9 0, [0, -1, 0, 2, 0], -100, -1 4, [0, -1, -1, 2, 0], -100, -16
評価2 - DQN 学習 80回
Counter({-100: 5,
269: 8,
164: 20,
301: 3,
211: 2,
235: 7,
105: 3,
372: 2,
333: 1,
238: 2,
316: 2,
196: 2,
253: 1,
242: 2,
312: 3,
343: 3,
340: 9,
179: 2,
267: 3,
270: 1,
74: 3,
328: 9,
309: 2,
284: 1,
137: 1,
285: 1,
360: 1,
374: 1})
200回学習した結果です。
評価2 - DQN 学習 200回
Counter({-100: 8,
238: 4,
340: 10,
0: 4,
267: 6,
106: 1,
74: 7,
105: 1,
328: 4,
235: 33,
309: 9,
228: 1,
270: 2,
32: 1,
301: 1,
196: 3,
341: 2,
269: 1,
374: 1,
372: 1})
学習時の報酬グラフは以下の通り、PPO のようにスムーズに学習が進んでおらず、DQN は本件に向いていないのかもしれません。
学習時の報酬グラフ - DQN 学習 200回
これは、報酬のクリッピング ※ に因るものかとも思いましたが、RLlib における報酬クリッピングの設定 clip_rewards はデフォルトで None であり、DQN のデフォルト設定 （dqn.py の DEFAULT_CONFIG） においても有効化しているような箇所は見当たりませんでした。
他の箇所で実施しているのかもしれませんが、今回は確認できませんでした。
※ 基本的には、 元の報酬の符号に合わせて 1, -1, 0 のいずれかに報酬の値が変更されてしまい、 報酬の大小は考慮されなくなる 本件であれば、 重量超過やマイナス個数のように報酬がマイナスにならない行動であれば どれでも良い事になってしまうと思われる
ちなみに、clip_rewards を True に設定した PPOTrainer を使ってみたところ、結果が著しく悪化しました。（マイナス報酬を避けるだけになった）
2. サンプル2 - sample2.ipynb
次に、行動によってのみ次の状態を決定するようにしてみます。 エピソードは 1回の行動で終了とします。
| 状態 | 行動 | （即時）報酬 |
|---|---|---|
| 品物毎の個数 | 品物毎の個数 | 価値の合計 |
この場合、action_space も Box で定義する事になります。
最軽量の品物が重要制限内で選択できる最大の個数を Box の最大値としています。
環境定義
def next_state(action): return [round(d) for d in action] def calc_reward(items, state, max_weight, burst_reward): reward = 0 weight = 0 for i in range(len(state)): reward += items[i][0] * state[i] weight += items[i][1] * state[i] if weight > max_weight or min(state) < 0: reward = burst_reward return reward, weight class Knapsack(gym.Env): def __init__(self, config): self.items = config["items"] self.max_weight = config["max_weight"] self.burst_reward = config["burst_reward"] # 個数の最大値 n = self.max_weight // min(np.array(self.items)[:, 1]) self.action_space = Box(0, n, shape = (len(self.items), )) self.observation_space = Box(0, n, shape = (len(self.items), )) def reset(self): return [0 for _ in self.items] def step(self, action): state = next_state(action) reward, _ = calc_reward(self.items, state, self.max_weight, self.burst_reward) return state, reward, True, {}
設定
items = [
[105, 10],
[74, 7],
[164, 15],
[32, 3],
[235, 22]
]
config = {
"env": Knapsack,
"env_config": {"items": items, "max_weight": 35, "burst_reward": -100}
}
(a) PPO（Proximal Policy Optimization）
PPO で実施してみます。
基本的な処理内容は 1. サンプル1 と同じですが、行動 1回でエピソードが終了するため、以下のコードで結果を確認する事にします。
評価
import collections rs = [] for _ in range(100): s = [0 for _ in range(len(items))] a = trainer.compute_action(s) s = next_state(a) r, _ = calc_reward(items, s, config["env_config"]["max_weight"], config["env_config"]["burst_reward"]) rs.append(r) collections.Counter(rs)
70回学習後の結果です。
評価結果 - PPO 学習 70回
Counter({375.0: 100})
学習時の状況は以下のようになりました。
学習時の報酬グラフ - PPO 学習 70回
(b) DDPG（Deep Deterministic Policy Gradient）
action_space が Box の場合に、DQNTrainer が UnsupportedSpaceException エラーを発生させるので、DQN は使えませんでした。
そこで、DDPG を使ってみました。
トレーナーの定義 - DDPG
from ray.rllib.agents.ddpg import DDPGTrainer trainer = DDPGTrainer(config = config)
こちらは、PPO 等と比べて処理に時間がかかる上に、80回学習しても順調とはいえない結果となりました。
学習時の報酬グラフ - DDPG 学習 80回
評価結果 - DDPG 学習 80回
Counter({347.0: 18, -100: 79, 242.0: 3})
以下のように評価の回数を 1000 にして再度確認してみます。
for _ in range(1000): s = [0 for _ in range(len(items))] a = trainer.compute_action(s) ・・・ collections.Counter(rs)
評価結果（1000回） - DDPG 学習 80回
Counter({347.0: 232,
-100: 718,
242.0: 24,
315.0: 14,
210.0: 1,
316.0: 7,
274.0: 3,
374.0: 1})
347（3, 0, 0, 1, 0）が多くなっているのはよく分かりませんが、DDPG も本件に向いていないのかもしれません。
Deno から npm パッケージを使用する（Deno で fp-ts）
下記の方法を用いて Node.js / ブラウザ用 npm パッケージを Deno から利用してみました。
- (a) Skypack の使用
- (b) Deno Node compatibility の使用
npm パッケージは関数プログラミング用 TypeScript ライブラリの fp-ts を試すことにします。
fp-ts は CommonJS と ES Modules のモジュール形式に対応していますが、現時点で Deno に対する直接的なサポートは無さそうでした。
また、使用した Deno のバージョンは以下の通りです。
- Deno 1.3.1
今回のサンプルコードは http://github.com/fits/try_samples/tree/master/blog/20200825/
はじめに
まずは、Node.js でサンプルコードを作成してみました。
sample.ts
import { Option, some, none, map } from 'fp-ts/lib/Option' import { pipe } from 'fp-ts/lib/pipeable' const f = (d: Option<number>) => pipe( d, map(v => v + 2), map(v => v * 3) ) console.log( f(some(5)) ) console.log( f(none) )
実行結果は以下の通りです。
sample.ts 実行結果
> npm install ts-node typescript fp-ts
・・・
> ts-node sample.ts
{ _tag: 'Some', value: 21 }
{ _tag: 'None' }
(a) Skypack の使用
前回の GraphQL.js でも利用しましたが、Skypack は npm パッケージをブラウザから直接使えるようにするための CDN となっています。
CommonJS 形式の npm パッケージを ES Modules 形式で提供する機能や Deno をサポートする機能（https://docs.skypack.dev/code/deno）が用意されているようです。
型情報なし
まずは、fp-ts の ES Modules のファイル（es6 に配置されている）を Skypack から import してみます。
TypeScript 用の型定義を指定しないと、関数の引数や戻り値などは any 型として扱う事になります。
また、import の際に拡張子を指定しなくても Skypack が .js ファイルを返してくれます。
a_1.ts
import { some, none, map } from 'https://cdn.skypack.dev/fp-ts/es6/Option.js' // 以下でも同じ //import { some, none, map } from 'https://cdn.skypack.dev/fp-ts/es6/Option' import { pipe } from 'https://cdn.skypack.dev/fp-ts/es6/pipeable.js' const f = (d: any) => // @ts-ignore pipe( d, map( (v: number) => v + 2 ), map( (v: number) => v * 3 ) ) console.log( f(some(5)) ) console.log( f(none) )
a_1.ts 実行結果
> deno run a_1.ts
・・・
{ _tag: "Some", value: 21 }
{ _tag: "None" }
ここで、@ts-ignore は Visual Studio Code におけるエラー表示対策 ※ のために付けています。
※ pipe 関数は、10個の any 型の引数をとる関数となっているが、
引数を 3つしか指定していない事に対するエラー
Visual Studio Code におけるエラー表示例（@ts-ignore を付けなかった場合）
なお、関数の引数や戻り値などを適切な型で扱うには、型定義ファイル（.d.ts）の指定が必要になりますが、fp-ts が用意している型定義ファイルを以下のように @deno-types で指定しても上手くいきません。
型定義ファイル指定の失敗例
// @deno-types="https://cdn.skypack.dev/fp-ts/es6/Option.d.ts" import { Option, some, none, map } from 'https://cdn.skypack.dev/fp-ts/es6/Option.js' // @deno-types="https://cdn.skypack.dev/fp-ts/es6/pipeable.d.ts" import { pipe } from 'https://cdn.skypack.dev/fp-ts/es6/pipeable.js' ・・・
というのも、fp-ts の Option.d.ts や pipeable.d.ts では import { ・・・ } from './Alt' のように .d.ts の拡張子を付けずに他の型定義を import しており、不都合が生じます。※
※ この場合、Skypack は Alt.js を返すことになり、
型情報を正しく取得できないと考えられる
ちなみに、Skypack 本来の使い方としては、以下のようにパッケージのルートを指定して import する事になりそうです。
a_2.ts
import { option, pipeable } from 'https://cdn.skypack.dev/fp-ts' const { some, none, map } = option const { pipe } = pipeable const f = (d: any) => // @ts-ignore pipe( d, map( (v: number) => v + 2 ), map( (v: number) => v * 3 ) ) console.log( f(some(5)) ) console.log( f(none) )
a_2.ts 実行結果
> deno run a_2.ts
・・・
{ _tag: "Some", value: 21 }
{ _tag: "None" }
型定義を自作
型に関しては、型定義ファイルを自作する事で一応は解決できます。
例えば、以下のような型定義ファイルを用意します。
types/Option.d.ts
export interface None { readonly _tag: 'None' } export interface Some<A> { readonly _tag: 'Some' readonly value: A } export declare type Option<A> = None | Some<A> export declare const some: <A>(a: A) => Option<A> export declare const none: Option<never> export declare const map: <A, B>(f: (a: A) => B) => (fa: Option<A>) => Option<B>
types/pipeable.d.ts
export declare function pipe<A, B, C, D, E, F, G, H, I, J>( a: A, ab: (a: A) => B, bc?: (b: B) => C, cd?: (c: C) => D, de?: (d: D) => E, ef?: (e: E) => F, fg?: (f: F) => G, gh?: (g: G) => H, hi?: (h: H) => I, ij?: (i: I) => J ): J
これを @deno-types で指定する事で型の問題が解決します。
a_3.ts
// @deno-types="./types/Option.d.ts" import { Option, some, none, map } from 'https://cdn.skypack.dev/fp-ts/es6/Option.js' // @deno-types="./types/pipeable.d.ts" import { pipe } from 'https://cdn.skypack.dev/fp-ts/es6/pipeable.js' const f = (d: Option<number>) => pipe( d, map( v => v + 2 ), map( v => v * 3 ) ) console.log( f(some(5)) ) console.log( f(none) )
a_3.ts 実行結果
> deno run a_3.ts
・・・
{ _tag: "Some", value: 21 }
{ _tag: "None" }
パッケージのルートを import する場合は、以下のような型定義ファイルを追加して @deno-types で指定します。
types/index.d.ts
import * as option from './Option.d.ts' import * as pipeable from './pipeable.d.ts' export { option, pipeable }
a_4.ts
// @deno-types="./types/index.d.ts" import { option, pipeable } from 'https://cdn.skypack.dev/fp-ts' const { some, none, map } = option const { pipe } = pipeable const f = (d: option.Option<number>) => pipe( d, map( v => v + 2 ), map( v => v * 3 ) ) console.log( f(some(5)) ) console.log( f(none) )
a_4.ts 実行結果
> deno run a_4.ts
・・・
{ _tag: "Some", value: 21 }
{ _tag: "None" }
dts クエリパラメータの利用
Skypack には、Deno 用に型定義ファイルを解決する手段として dts クエリパラメータが用意されています。
これを使う事で、本来は以下のようなコードで型問題を解決できるはずですが、fp-ts 2.8.2 では上手くいきませんでした。
a_5e.ts
import { option, pipeable } from 'https://cdn.skypack.dev/fp-ts?dts' const { some, none, map } = option const { pipe } = pipeable const f = (d: option.Option<number>) => pipe( d, map( v => v + 2 ), map( v => v * 3 ) ) console.log( f(some(5)) ) console.log( f(none) )
実行結果は以下のようになり、型定義ファイルの取得途中で 404 Not Found エラーが発生してしまいます。
a_5e.ts 実行結果
> deno run a_5e.ts ・・・ Download https://cdn.skypack.dev/-/fp-ts@v2.8.2-Hr9OPgW5wz4u6TqOfiZH/dist=es2020,mode=types/lib/TaskEither.d.ts error: Import 'https://cdn.skypack.dev/-/fp-ts@v2.8.2-Hr9OPgW5wz4u6TqOfiZH/dist=es2020,mode=types/lib/HKT.d.ts' failed: 404 Not Found Imported from "https://cdn.skypack.dev/-/fp-ts@v2.8.2-Hr9OPgW5wz4u6TqOfiZH/dist=es2020,mode=types/lib/index.d.ts:42"
これは、fp-ts の中で HKT だけ特殊な扱いがされており、HKT.d.ts ファイルが lib ディレクトリ内に配置されておらず、パッケージのルートディレクトリに配置されている事が原因だと考えられます。※
※ そのため、
"/lib/HKT.d.ts" ではなく "/HKT.d.ts" を import する必要がある
Node.js においては
"/lib/HKT/package.json" の typings フィールドの値から
HKT.d.ts の配置場所を取得するようになっていると思われるが、
Skypack の dts クエリパラメータの機能では
そこまでを考慮していない事が原因だと考えられる
ここで、dts クエリパラメータによって何が変わるのかというと、?dts を付けることでレスポンスヘッダーに x-typescript-types が付与され、型定義ファイル（.d.ts）の取得先が提示されるようになります。※
※ Deno は x-typescript-types ヘッダーの値から
型定義ファイルを自動的に取得するようになっている
dts クエリパラメータの有無による違い
$ curl --head -s https://cdn.skypack.dev/fp-ts | grep x-typescript-types
$ curl --head -s https://cdn.skypack.dev/fp-ts?dts | grep x-typescript-types x-typescript-types: /-/fp-ts@v2.8.2-Hr9OPgW5wz4u6TqOfiZH/dist=es2020,mode=types/lib/index.d.ts
更に、x-typescript-types のパスから取得できる index.d.ts は、fp-ts のオリジナル index.d.ts を（Deno 用に）加工したもの ※ となっています。
※ 型定義ファイル内の
import 対象のパスに拡張子 .d.ts を加えている
fp-ts の HKT で起きた問題を考えると、
現時点では .d.ts ファイルの存在有無や
package.json の typings フィールド等の考慮はされていないと考えられる
オリジナルの index.d.ts 内容
・・・ import * as alt from './Alt' ・・・
x-typescript-types の index.d.ts 内容
・・・ import * as alt from './Alt.d.ts' ・・・
ついでに GraphQL.js に対して確認してみましたが、こちらは（npm パッケージ内に .d.ts ファイルが存在するものの） ?dts を付けても x-typescript-types ヘッダーは付与されませんでした。
GraphQL.js の場合
$ curl --head -s https://cdn.skypack.dev/graphql?dts | grep x-typescript-types
(b) Deno Node compatibility の使用
Deno Node compatibility は Deno の std ライブラリに含まれている機能で CommonJS モジュールのロードだけではなく、Node.js API との互換 API もある程度は用意されているようです。（fs.readFile 等）
ただ、require の戻り値の型は any となってしまうので、TypeScript で使うメリットはあまり無いかもしれません。
b_1.ts
import { createRequire } from 'https://deno.land/std/node/module.ts' const require = createRequire(import.meta.url) const { some, none, map } = require('fp-ts/lib/Option') const { pipe } = require('fp-ts/lib/pipeable') const f = (d: any) => pipe( d, map( (v: number) => v + 2), map( (v: number) => v * 3) ) console.log( f(some(5)) ) console.log( f(none) )
fp-ts を npm でインストールしてから deno run で実行します。 deno run で実行するには下記オプションを指定する必要がありました。
- --unstable
- --allow-read
- --allow-env
b_1.ts 実行結果
> npm install fp-ts
・・・
> deno run --unstable --allow-read --allow-env b_1.ts
・・・
{ _tag: "Some", value: 21 }
{ _tag: "None" }
