gotoshin

主に学んだ事の自分メモ用です。記事に書くまでも無いような事はhttps://scrapbox.io/study-diary/に書いてます。

grpcコースをやってみる〜その4 part1 ユーザ作成を実装する〜

www.youtube.com

今回何をやるか

未実装のAPIを実装する

実装する

完成形のコードは以下

func (server *Server) CreateUser(ctx context.Context, req *pb.CreateUserRequest) (*pb.CreateUserResponse, error) {
    hashedPassword, err := util.HashPassword(req.GetPassword())
    if err != nil {
        return nil, status.Errorf(codes.Internal, "failed to hash password: %s", err)
    }

    arg := db.CreateUserParams{
        Username:       req.GetUsername(),
        HashedPassword: hashedPassword,
        FullName:       req.GetFullName(),
        Email:          req.GetEmail(),
    }

    user, err := server.store.CreateUser(ctx, arg)
    if err != nil {
        if pqErr, ok := err.(*pq.Error); ok {
            errName := pqErr.Code.Name()
            switch errName {
            case "unique_violation":
                return nil, status.Errorf(codes.AlreadyExists, "username already exists: %s", err)
            }
        }
        return nil, status.Errorf(codes.Internal, "failed to create user: %s", err)
    }

    rsp := &pb.CreateUserResponse{
        User: convertUser(user),
    }

    return rsp, nil
}

以下に実装内容の補足を書いていく

リクエストパラメータのバインドが不要

ginでは以下の様にパラメータをバインドする必要があったが、grpcではフレームワークで実施してくれるためバインドが不要

// api/user.go

// jsonのキー名とマッピングさせたリクエスト
type createUserRequest struct {
    Username string `json:"username" binding:"required,alphanum"`
    Password string `json:"password" binding:"required,min=6"`
    FullName string `json:"full_name" binding:"required"`
    Email    string `json:"email" binding:"required,email"`
}

// メソッド
func (server *Server) createUser(ctx *gin.Context) {
    // jsonとバインドできているか検査
    var req createUserRequest
    if err := ctx.ShouldBindJSON(&req); err != nil {
        ctx.JSON(http.StatusBadRequest, errorResponse(err))
        return
    }

grpcのエラーハンドリング方法

httpの場合jsonにhttpのステータスコードを設定していたが、grpcの場合status（grpcのサブパッケージ）にエラーコードを設定する

// http
ctx.JSON(http.StatusInternalServerError, errorResponse(err))

// grpc
return nil, status.Errorf(codes.Unimplemented, "method CreateUser not implemented")

// 呼び出し先
// Errorf returns Error(c, fmt.Sprintf(format, a...)).
func Errorf(c codes.Code, format string, a ...interface{}) error {
    return Error(c, fmt.Sprintf(format, a...))
}

dbで生成したデータをproto型へ変換する

  • dbで生成したユーザを直接クライアントへ返却しないのが一般的である
    • ユーザテーブルのフィールドをそのまま返却した場合パスワード情報など不要でかつ公開したくない情報まで返却することになるため
// レスポンスとして使用したいデータ型
// pb/rpc_create_user.pb.go = protoで自動生成したデータ型
type CreateUserResponse struct {
    state         protoimpl.MessageState
    sizeCache     protoimpl.SizeCache
    unknownFields protoimpl.UnknownFields

    User *User `protobuf:"bytes,1,opt,name=user,proto3" json:"user,omitempty"`
}

以下の様に変換用の処理を作成し変換する

// gapi/converter.go
func convertUser(user db.User) *pb.User {
    return &pb.User{
        Username:          user.Username,
        FullName:          user.FullName,
        Email:             user.Email,
        PasswordChangedAt: timestamppb.New(user.PasswordChangedAt),
        CreatedAt:         timestamppb.New(user.CreatedAt),
    }
}

   // レスポンス生成処理 抜粋
    rsp := &pb.CreateUserResponse{
        // 引数で受け取っているユーザはdbで生成したユーザ 
        User: convertUser(user),
    }

    return rsp, nil

[Go] 構造体の初期化方法まとめ - Qiita

実際に動作させてみる

evansから呼び出し方法を確認したところ成功！

pb.SimpleBank@localhost:9090> call CreateUser
username (TYPE_STRING) => a
full_name (TYPE_STRING) => a
email (TYPE_STRING) => a@gmail.com
password (TYPE_STRING) => password
{
  "user": {
    "createdAt": "2023-05-05T01:49:38.907170Z",
    "email": "a@gmail.com",
    "fullName": "a",
    "passwordChangedAt": "0001-01-01T00:00:00Z",
    "username": "a"
  }
}

感想

  • これまでgrpcの概要から入り実際に呼び出すところまでやったが、httpと手続きが異なるだけで呼び出し後の処理（処理を書く、エラーハンドリングをする・・・etc）などは同じだなという感覚。

grpcコースをやってみる〜その3〜

www.youtube.com

今回何をやるか

hatehate-nazenaze.hatenablog.com

前回はデータスキーマとそれを利用したapiを定義したが、今回はそれを動作させるサーバを定義する

serverの雛形を作る

  • api/sever.goのコードの中身をコピペする
  • 異なるのはhttpリクエストではなく、grpcリクエストを取り扱う点
  • フィールドであるconfig、store、tokenMakerは引き続き必要である
  • 以下は削除する
    • ginで動作するValidatorはgrpcでは動作しないため削除
    • routerのセットアップもgrpcにはルートはないため削除
      • クライアントはrpcを利用してあたかもローカルの関数を呼び出すかのように処理を呼び出す
// gapi/server.go
type Server struct {
    config     util.Config
    store      db.Store
    tokenMaker token.Maker
}

func NewServer(config util.Config, store db.Store) (*Server, error) {
    tokenMaker, err := token.NewPasetoMaker(config.TokenSymmetricKey)
    if err != nil {
        return nil, fmt.Errorf("cannot create token maker: %w", err)
    }
    server := &Server{
        config:     config,
        store:      store,
        tokenMaker: tokenMaker,
    }

    return server, nil
}

serverが必要なインターフェースを継承する様に修正する

grpcサーバとして振る舞うには、自動生成された以下のインターフェースを継承する必要がある

// pb/service_simple_bank_grpc.pb.go
type SimpleBankServer interface {
    CreateUser(context.Context, *CreateUserRequest) (*CreateUserResponse, error)
    LoginUser(context.Context, *LoginUserRequest) (*LoginUserResponse, error)
    mustEmbedUnimplementedSimpleBankServer()
}

以下のようにmustEmbedUnimplementedSimpleBankServerメソッドを持っているUnimplementedSimpleBankServerをフィールドに追加する

type Server struct {
        // 追加
    pb.UnimplementedSimpleBankServer
    config     util.Config
    store      db.Store
    tokenMaker token.Maker
}

mustEmbedUnimplementedSimpleBankServer()とは？

  • UnimplementedSimpleBankServerをフィールドとして持つことにより、サーバは未実装のメソッドを持つことができる
    • IFだけを定義した状態で徐々に実装していくことが可能、チーム開発などで便利そう
  • 定義内容は以下の通り
// pb/service_simple_bank_grpc.pb.go
// UnimplementedSimpleBankServer must be embedded to have forward compatible implementations.
type UnimplementedSimpleBankServer struct {
}

func (UnimplementedSimpleBankServer) CreateUser(context.Context, *CreateUserRequest) (*CreateUserResponse, error) {
    return nil, status.Errorf(codes.Unimplemented, "method CreateUser not implemented")
}
func (UnimplementedSimpleBankServer) LoginUser(context.Context, *LoginUserRequest) (*LoginUserResponse, error) {
    return nil, status.Errorf(codes.Unimplemented, "method LoginUser not implemented")
}
func (UnimplementedSimpleBankServer) mustEmbedUnimplementedSimpleBankServer() {}

serverの呼び出し元であるmain.goファイルを修正する

現状以下のようにginでhttpサーバを構築している

// main/main.go
func main() {
    config, err := util.LoadConfig(".")
    if err != nil {
        log.Fatal("cannot load config:", err)
    }

    conn, err := sql.Open(config.DBDriver, config.DBSource)
    if err != nil {
        log.Fatal("cannot connect to db:", err)
    }

    store := db.NewStore(conn)
    server, err := api.NewServer(config, store)
    if err != nil {
        log.Fatal("cannot create server:", err)
    }

    err = server.Start(config.ServerAddress)
    if err != nil {
        log.Fatal("cannot start server:", err)
    }
}

これを以下の様にすることで、grpcとhttpを切り替えやすくする ※動画の都合上、httpサーバも残しておきたいとのこと

// main/main.go
func main() {
    config, err := util.LoadConfig(".")
    if err != nil {
        log.Fatal("cannot load config:", err)
    }

    conn, err := sql.Open(config.DBDriver, config.DBSource)
    if err != nil {
        log.Fatal("cannot connect to db:", err)
    }

    store := db.NewStore(conn)
    runGinServer(config, store)
}

  // ここに今回の処理を追記していく
func runGrpcServer(config util.Config, store db.Store) {
}

func runGinServer(config util.Config, store db.Store) {
    server, err := api.NewServer(config, store)
    if err != nil {
        log.Fatal("cannot create server:", err)
    }

    err = server.Start(config.ServerAddress)
    if err != nil {
        log.Fatal("cannot start server:", err)
    }
}

main/main.go runGrpcServerメソッドの中身を実装する

今回作成したサーバをgrpcサーバに対して登録する

// main/main.go
func runGrpcServer(config util.Config, store db.Store) {
    server, err := gapi.NewServer(config, store)
    if err != nil {
        log.Fatal("cannot create server:", err)
    }

    grpcServer := grpc.NewServer()
    pb.RegisterSimpleBankServer(grpcServer, server)
}

grpc.NewServer()

サービス未登録で、リクエストを受け付けていないサーバを作成する

メソッドのコメントは以下のような記載になっている

// grpc@v1.43.0/server.go

// NewServer creates a gRPC server which has no service registered and has not
// started to accept requests yet.

RegisterSimpleBankServer()

引数で受け取ったgrpcサーバに対して、サービスの説明とサービスを登録

// pb/service_simple_bank_grpc.pb.go
func RegisterSimpleBankServer(s grpc.ServiceRegistrar, srv SimpleBankServer) {
    s.RegisterService(&SimpleBank_ServiceDesc, srv)}

s.RegisterService()

// ServiceRegistrar wraps a single method that supports service registration. It
// enables users to pass concrete types other than grpc.Server to the service
// registration methods exported by the IDL generated code.
type ServiceRegistrar interface {
    // RegisterService registers a service and its implementation to the
    // concrete type implementing this interface.  It may not be called
    // once the server has started serving.
    // desc describes the service and its methods and handlers. impl is the
    // service implementation which is passed to the method handlers.
    RegisterService(desc *ServiceDesc, impl interface{})
}

SimpleBank_ServiceDesc

サービスに関する説明のよう

// pb/service_simple_bank_grpc.pb.go

// SimpleBank_ServiceDesc is the grpc.ServiceDesc for SimpleBank service.
// It's only intended for direct use with grpc.RegisterService,
// and not to be introspected or modified (even as a copy)
var SimpleBank_ServiceDesc = grpc.ServiceDesc{
    ServiceName: "pb.SimpleBank",
    HandlerType: (*SimpleBankServer)(nil),
    Methods: []grpc.MethodDesc{
        {
            MethodName: "CreateUser",
            Handler:    _SimpleBank_CreateUser_Handler,
        },
        {
            MethodName: "LoginUser",
            Handler:    _SimpleBank_LoginUser_Handler,
        },
    },
    Streams:  []grpc.StreamDesc{},
    Metadata: "service_simple_bank.proto",
}

reflection.Register(grpcServer)を追記する

この一文を追加することで、クライアントが容易にサービスの一覧や、型や、詳細を取得する事が出来る

grpc-goのリフレクションについて調べてみた - Qiita

// main/main.go
func runGrpcServer(config util.Config, store db.Store) {
    server, err := gapi.NewServer(config, store)
    if err != nil {
        log.Fatal("cannot create server:", err)
    }

    grpcServer := grpc.NewServer()
    pb.RegisterSimpleBankServer(grpcServer, server)
    // 追記
    reflection.Register(grpcServer)
}

設定を編集する

grpcサーバのポートを追加する

環境定義ファイルにgrpcサーバのポートを追加する

DB_DRIVER=postgres
DB_SOURCE=postgresql://root:secret@localhost:5432/simple_bank?sslmode=disable
HTTP_SEVER_ADDRESS=0.0.0.0:8080
GRPC_SEVER_ADDRESS=0.0.0.0:9090
TOKEN_SYMMETRIC_KEY=12345678901234567890123456789012
ACCESS_TOKEN_DURATION=15m

合わせてマッピング先のデータ構造を修正する

// config.go
type Config struct {
    DBDriver            string        `mapstructure:"DB_DRIVER"`
    DBSource            string        `mapstructure:"DB_SOURCE"`
    HttpServerAddress   string        `mapstructure:"HTTP_SEVER_ADDRESS"`
    GrpcServerAddress   string        `mapstructure:"GRPC_SEVER_ADDRESS"`
    TokenSymmetricKey   string        `mapstructure:"TOKEN_SYMMETRIC_KEY"`
    AccessTokenDuration time.Duration `mapstructure:"ACCESS_TOKEN_DURATION"`
}

リクエスト受付処理を追記する

// main.go
func runGrpcServer(config util.Config, store db.Store) {
    server, err := gapi.NewServer(config, store)
    if err != nil {
        log.Fatal("cannot create server:", err)
    }

    grpcServer := grpc.NewServer()
    pb.RegisterSimpleBankServer(grpcServer, server)
    reflection.Register(grpcServer)
    // 追記
    listener, err := net.Listen("tcp", config.GrpcServerAddress)
    if err != nil {
        log.Fatal("cannot create listener", err)
    }

    log.Printf("start gRPC server at %s", listener.Addr().String())
    err = grpcServer.Serve(listener)
    if err != nil {
        log.Fatal("cannot start gRPC server")
    }
}

listener, err := net.Listen("tcp", config.GrpcServerAddress)

コネクションを取得することらしい ネットワーク｜Goから学ぶI/O

grpcServer.Serve(listener)

サービスを登録した状態のgrpcServerに対して、コネクションを渡してリクエストの受付を開始すると理解した

// Serve accepts incoming connections on the listener lis, creating a new
// ServerTransport and service goroutine for each. The service goroutines
// read gRPC requests and then call the registered handlers to reply to them.
// Serve returns when lis.Accept fails with fatal errors.  lis will be closed when
// this method returns.
// Serve will return a non-nil error unless Stop or GracefulStop is called.
func (s *Server) Serve(lis net.Listener) error {

実際に動作させてみる

// main.go
ffunc main() {
    config, err := util.LoadConfig(".")
    if err != nil {
        log.Fatal("cannot load config:", err)
    }

    conn, err := sql.Open(config.DBDriver, config.DBSource)
    if err != nil {
        log.Fatal("cannot connect to db:", err)
    }

    store := db.NewStore(conn)
         // grpcサーバ起動へ変更
    runGrpcServer(config, store)
}

起動

以下の通り起動自体は成功する

 ~/develop/simple-bank-cp | master +2 !4  make server                                                       INT | 4s
go run main.go
2023/04/28 23:32:56 start gRPC server at [::]:9090

動作確認

Evansというgrpc clientを利用する

github.com

brew tapとは - Qiita

evans -r repl

If your server is enabling gRPC reflection, you can launch Evans with only -r (--reflection) option. gRPC reflectionが有効になっている場合、-rオプションをつけることで起動できるらしい そのままやるとgRPCのデフォルトポートである50051に接続しにいくため、エラーとなる 以下の通りオプションを追加することで接続に成功する

evans --host localhost --port 9090 -r repl
 ~/develop/simple-bank-cp | master +2 !4  evans --host localhost --port 9090 -r repl                           1 err

  ______
 |  ____|
 | |__    __   __   __ _   _ __    ___
 |  __|   \ \ / /  / _. | | '_ \  / __|
 | |____   \ V /  | (_| | | | | | \__ \
 |______|   \_/    \__,_| |_| |_| |___/

 more expressive universal gRPC client


pb.SimpleBank@localhost:9090>

繋いだ状態でいくつかリクエストを試す事が可能。 ※まだ未実装のため、エラーとなる

grpcコースをやってみる〜その2〜

www.youtube.com

環境構築

Protocol Buffer Compilerをinstall

Protocol Buffer Compiler Installation | gRPC

記載の通りbrew installしたらいけた

Goをビルドするためにプラグインをinstall

protoc-gen-go

prot ファイルで定義されたデータからgoのリクエスト・レスポンスデータを生成する

pkg.go.dev

protoc-gen-go-grpc

grpcフレームワークで動くgoのコードを生成する

pkg.go.dev

これで準備は完了

ユーザ新規作成のgrpc APIを作成

protoスキーマを定義する

ユーザのスキーマ

作成元のデータ

ユーザ新規作成時のレスポンスを元データとする

// /api/user.go
type userResponse struct {
    Username          string    `json:"username"`
    FullName          string    `json:"full_name"`
    Email             string    `json:"email"`
    PasswordChangedAt time.Time `json:"password_changed_at"`
    CreatedAt         time.Time `json:"created_at"`
}

proto定義

// proto/user.proto

option go_package = "github.com/simplebank/pb";

// シンタックス
syntax = "proto3";

// パッケージ
package pb;

import "google/protobuf/timestamp.proto";

// どのパッケージに対してgoコードを生成するか
// 下の記述によってpbフォルダ配下に生成される（フォルダは事前作成する必要あり）
option go_package = "github.com/simplebank/pb";

message User {
  string username = 1;
  string full_name = 2;
  string email = 3;
  // デフォルトのprotoにはタイムスタンプ型は無いため、ライブラリを利用する
  google.protobuf.Timestamp password_changed_at =4;
  google.protobuf.Timestamp created_at=5;
}

リクエスト・レスポンスのスキーマ

作成元のデータ

ユーザ作成のリクエスト

type createUserRequest struct {
    Username string `json:"username" binding:"required,alphanum"`
    Password string `json:"password" binding:"required,min=6"`
    FullName string `json:"full_name" binding:"required"`
    Email    string `json:"email" binding:"required,email"`
}

proto定義

syntax = "proto3";

package pb;

import "user.proto";

option go_package = "github.com/simplebank/pb";

message CreateUserRequest {
  string username = 1;
  string full_name = 2;
  string email = 3;
  string password = 4;
}

// レスポンスは最初に作成したユーザ定義をそのまま返却する
message CreateUserResponse {
  User user = 1;
}

grpc APIを定義する

syntax = "proto3";

package pb;

// 先ほど定義したスキーマを読み込み
import "rpc_create_user.proto";

option go_package = "github.com/simplebank/pb";

service SimpleBank {
  rpc CreateUser (CreateUserRequest) returns (CreateUserResponse) {}
}

これにてユーザ作成のAPIは完成

詰まった箇所

問題

  1. 通常通りクラスを記載しても型を参照出来ないため、import文を追加する
  2. import文が赤線でエラー表記となる

対策

解決するためには、拡張機能の説明に記載のある内容を参考にsetting.jsonに以下の内容を追記

  "protoc": {
      "options": [
          "--proto_path=protos",
      ]
  }

VSCodeを再起動して解消

同じ流れでログインユーザのgrpc APIを作成

通信タイプは？

シンプルなunaryで実施。

protoスキーマを定義する

リクエスト・レスポンスデータ

作成元のデータ

動画のものと比べて古い。

type loginUserRequest struct {
    Username string `json:"username" binding:"required,alphanum"`
    Password string `json:"password" binding:"required,min=6"`
}

type loginUserResponse struct {
    AccessToken string       `json:"access_token"`
    User        userResponse `json:"user"`
}
  • userResponseクラスは最初に定義したUserスキーマと同じため、流用可能

proto定義

syntax = "proto3";

package pb;

import "user.proto";

option go_package = "github.com/simplebank/pb";

message LoginUserRequest {
  string username = 1;
  string password = 2;
}

message LoginUserResponse {
  string access_token= 1;
  User user = 2;
}

grpc APIを定義する

service SimpleBank {
  rpc CreateUser (CreateUserRequest) returns (CreateUserResponse) {}
  // ログイン用に追加
  rpc LoginUser (LoginUserRequest) returns (LoginUserResponse) {}
}

go コードを生成する

grpc.io 公式からgoコード生成用のコマンドを引っ張ってきてMakefileに追記する

// Makefile
proto:
    protoc --proto_path=proto --go_out=pb --go_opt=paths=source_relative \
    --go-grpc_out=pb --go-grpc_opt=paths=source_relative \
    proto/*.proto
・
・
・
.PHONY: postgres createdb dropdb migrateup migratedown migrateup1 migratedown1 sqlc server mock proto

ターミナル上でmake protoを実行することでgoのコードが生成される

  • それぞれのファイルは今回作成したprotoファイルを元に生成されている

  • pb/sevice_simple_bank_grpc.pb.goはサーバとクライアントのインターフェースとスタブを含んでいる

パッケージimoprtエラーを解消する

  • 単純にgoogle.golang.org/grpcパッケージが入っていない事が原因
  • go mod tidyを実施することで不足分が入ってくる

Makefileを修正する

// Makefile
proto:
        // 追加
    rm -f pb/*go
    protoc --proto_path=proto --go_out=pb --go_opt=paths=source_relative \
    --go-grpc_out=pb --go-grpc_opt=paths=source_relative \
    proto/*.proto

生成都度ファイルが増えることを防ぐため、生成前に削除するコマンドを追加する

grpcコースをやってみる〜その1〜

今回やってみるコース

www.youtube.com

モチベーション

仕事で触れる機会があるが仕組みが何となくしかわかっていないため手を動かしながらやってみる

grpcとは

  • クライアントがリモートサーバの関数を呼び出す
    • 例えばRustで記述されたサーバの関数をGoのローカルから呼び出すことが可能
  • APIとデータ構造はProtocol Buffer Compilerによって自動生成される
  • 様々な異なる言語をサポートしている

どのように動くか

  1. protbufを使用しデータ構造とAPIの定義を作成する
  2. 作成した定義を元にサーバとクライアントのコードを自動生成する
    1. Protocol Buffer Compilerによって、選択した言語のコードを簡単に自動生成する事が可能
  3. 最後にサーバとクライアント側で呼び出し側と呼び出し元の実装をする

なぜHTTP JSON API ではなくGRPCを利用するのか

高パフォーマンスである

  1. HTTP2を利用しているため
  2. 転送データをバイナリフォーマットにする
  3. 複数リクエストを同じTCPコネクションで転送する
    1. http2 - What does multiplexing mean in HTTP/2 - Stack Overflow
  4. ヘッダ圧縮
    1. 今更ながら HTTP/2 について調べてまとめてみた
  5. 双方向通信
    1. HTTP/2における双方向通信とgRPCとこれから - Qiita

Strong API Contract

サーバとクライアントは同じprotを利用するため、データ構造などを同じものを利用できる（というか強制される）

コードの自動生成

  • サービス開発をより早くしてくれる
  • クライアント・サーバで利用するコードはgrpcによって自動生成されたコードを利用するだけなので、開発者はロジックに集中できる

4タイプの通信方法

grpc gateway

  • 1度のAPIコードの記載でhttpコードとgrpc両方を対応可能にする
  • 仕組みはhttpリクエストを一度gatewayを介すことによりgrpcリクエストへ変換し内部的にgrpcで処理する。レスポンス時はその逆。

【PostgreSQL】複数トランザクションを並列実行時の順序制限によるdeadlock回避

この記事について

前回の動画ではクエリの機能を活用しdeadlockを防いでいたが、今回は実行するクエリの順序を調整してデータロックを防ぐ

How to avoid deadlock in DB transaction? Queries order matters! www.youtube.com

事象

二つのトランザクション内で、同じレコードのデータを異なる順序で更新しようとすると、お互いのCommit待ちとなりdeadlockが発生してしまう

  • 以下の処理が同時に実行された場合、
    • 口座1⇨口座2へ10ドル送金
    • 口座2⇨口座1へ10ドル送金

① Transaction1内で口座1を10ドルプラス⇨成功

// Transaction1を開始
BEGIN;
UPDATE accounts SET balance = balance + 10 WHERE id = 1 RETURNING *;

② Transaction2内で口座2を10ドルプラス⇨成功

// Transaction2を開始
BEGIN;
UPDATE accounts SET balance = balance + 10 WHERE id = 2 RETURNING *;

③ Transaction1内で口座2を10ドルマイナス⇨口座2はTransaction2内で更新中のためロック

// Transaction1内
UPDATE accounts SET balance = balance - 10 WHERE id = 2 RETURNING *;

④ Transaction2内で口座1を10ドルマイナス⇨口座1はTransaction1内で更新中のためロック

// Transaction2内
UPDATE accounts SET balance = balance - 10 WHERE id = 1 RETURNING *;

deadlock発生！！！

解決策

更新する順序を揃えることで、片方のトランザクションのクエリがCommitされるまでもう片方のトランザクションのクエリは実行不可となり、deadlockは発生しない

① Transaction1内で口座1を10ドルプラス⇨成功

// Transaction1を開始
BEGIN;
UPDATE accounts SET balance = balance + 10 WHERE id = 1 RETURNING *;

② Transaction2内で口座1を10ドルマイナス⇨口座1はTransaction1内で更新中のためロック

// Transaction2を開始
BEGIN;
UPDATE accounts SET balance = balance + 10 WHERE id = 2 RETURNING *;

③ Transaction1内で口座2を10ドルマイナス⇨成功しTransaction1はCommitされる、その後中止されていた②の処理が動く

// Transaction1内
UPDATE accounts SET balance = balance - 10 WHERE id = 2 RETURNING *;

④ Transaction2内で口座2を10ドルプラス⇨成功

// Transaction2内
UPDATE accounts SET balance = balance - 10 WHERE id = 1 RETURNING *;

【PostgreSQL】 並列実行時のトランザクション管理の注意が必要なパターンと解決策

この記事について

前提

  • ある口座から別の口座へ金額を移すAPIを2つのゴールチンでテストして、トランザクションの正当性を確認している

同じ時点のデータを2ルーチンが同時に更新してしまうパターン

事象

  • 更新前
    • 送金元:380
    • 送金先:390
      • 送金元から送金先へ10ドル送付するという処理を2ルーチンで動かす
  • 更新後
    • ルーチン1
      • 送金元:370（380-10）
      • 送金先:400（390+10）
        • 想定通り
    • ルーチン2
      • 送金元370 ※想定360
      • 送金先410（400+10）
        • 送金元が想定と異なる！

原因

  • select時に2ルーチンで同じ時点（値が370時点）の送信元データを取得してしまっているため

解決策

  • SELECT文にFOR UPDATEを追加することで、あるルーチンでデータ取得している際は、そのルーチンが COMMIT か ROLLBACK するまでブロックされ処理が止まる
SELECT * FROM accounts 
WHERE id = $1 LIMIT 1
FOR UPDATE;

外部キー制約が原因でクエリの実行がブロックされるパターン

事象

  • 更新用にaccounts取得のクエリを実行したところ、transfers tableのレコード 挿入が原因でクエリの実行がブロックされる
// ルーチン2で実行し、ブロックされたクエリ
SELECT * FROM accounts 
WHERE id = $1 LIMIT 1
FOR UPDATE;

// ルーチン1で実行し、ブロック原因になっているクエリ
INSERT INTO transfers (
    from_account_id,
    to_account_id,
    amount
) VALUES (
    $1,
    $2,
    $3
) RETURNING *;

原因

  • 前提として、transfers tableのfrom_account_idと to_account_idは、accountstableの外部キーとなっている。
  • transfers tableの更新が行われるトランザクション実行中に、万が一accounts tableのidが書き換えられてしまうと、不整合なtrasferレコードを作成してしまうことになる。
  • それを防ぐためtransfers tableの実行中はaccountstableの更新ができない
  • 動画ではルーチン1が更に SELECT * FROM accounts・・・を発行し、ルーチン2でそれ以前に実行されたINSERT INTO transfersによりブロック、一方でルーチン2はルーチン1によってブロックされているという状態つまりdeadlock状態となっていた

解決策

  • SELECT文にFOR NO KEY UPDATE を追加することで、キーの更新が行われないことをPostgreSQLへ伝える事ができ、ブロックされなくなる
SELECT * FROM accounts 
WHERE id = $1 LIMIT 1
FOR NO KEY UPDATE;

後日追記

Play Frameworkを使ったWebアプリケーション作成 掲示板作成

ドワンゴ社 新卒エンジニア向けのScala研修資料をやっていくもの ※テスト項目の手前の段階でページが正常に表示される事を確認する

資料

Introduction · Scala研修テキスト

投稿されたメッセージのクラスを作る

package controllers

import java.time.OffsetDateTime

case class Post(body: String, date: OffsetDateTime)

ケースクラスとは

いつ使うのか

  • もし、イミュータブルなオブジェクトを使って純粋に関数的なコードを書こうとするならば、通常のクラスは使わないようにした方がよいでしょう。関数型パラダイムの主な考え方は、データ構造とそれに対する操作を分離することです。ケースクラスは、必要なメソッドを持つデータ構造の表現です。データに対する関数は、別のソフトウェアエンティティ (例: traits、object) で記述されるべきです。 これに対し、通常のクラスは、データと操作を結びつけて、ミュータビリティを提供する。この方法は、オブジェクト指向のパラダイムに近い。

な、なるほど〜！ 今回でいうと画面からのリクエストメッセージはイミュータブルなデータとして扱うから、caseクラスを使用していると理解した。

投稿されたメッセージの保存と取得

package controllers

object PostRepository {

  var posts: Seq[Post] = Vector()

  def findAll: Seq[Post] = posts

  def add(post: Post): Unit = { posts = posts :+ post }
}

objectで定義されている理由

自分の理解だが、関数のみを定義しフィールドを保持しないため

クラス名のsufixがRepositoryの理由

var posts: Seq[Post] = Vector

Vector
  • コレクションクラスの一つで、色んな操作において速度が安定しており、まず最初に検討すると良いクラスの一つだそうな
  • imutableなので一度データ構造を構築したら変更出来ない
    • と言いつつVector(1, 2, 3, 4, 5).updated(2, 5)みたいな例が書いてあり、更新出来そうに見える？
    • 自分の頭の中の可変・不変の定義が誤っていた...、変えてそうに見えて別コレクションを返しているそう
      • 可変コレクションおよび不変コレクション | Collections | Scala Documentation

        • Scala のコレクションは、体系的に可変および不変コレクションを区別している。可変 (mutable) コレクションは上書きしたり拡張することができる。これは副作用としてコレクションの要素を変更、追加、または削除することができることを意味する。一方、不変 (immutable) コレクションは変わることが無い。追加、削除、または更新を模倣した演算は提供されるが、全ての場合において演算は新しいコレクションを返し、古いコレクションは変わることがない。

Seq

def findAll: Seq[Post] = posts

  • 通常であればここはDBから取得した値を返すが、今回は変数を返す仕様となっている

def add(post: Post): Unit = { posts = posts :+ post }

  • 変数に対して処理を行なっている理由は上の記載と同様
Unit型とは

コントローラ

※私の環境はplay 2.8を使用したため、資料の表記と異なる

package controllers

import java.time.OffsetDateTime
import javax.inject.Inject
import play.api.data.Form
import play.api.data.Forms._
import play.api.i18n.I18nSupport
import play.api.mvc.{AbstractController, ControllerComponents}

case class PostRequest(body: String)

class TextboardController @Inject()(cc: ControllerComponents) extends AbstractController(cc) with I18nSupport {
  private[this] val form = Form(
    mapping(
      "post" -> text(minLength = 1, maxLength = 10).withPrefix("hogeika")
    )(PostRequest.apply)(PostRequest.unapply))

  def get = Action { implicit request =>
    Ok(views.html.index(PostRepository.findAll, form))
  }

  def post = Action { implicit request =>
    form.bindFromRequest().fold(
      error => BadRequest(views.html.index(PostRepository.findAll, error)),
      postRequest => {
        val post = Post(postRequest.body, OffsetDateTime.now)
        PostRepository.add(post)
        Redirect("/")
      }
    )
  }
}

PlayのForm

  private[this] val form = Form(
    mapping(
      "post" -> text(minLength = 1, maxLength = 10)
    )(PostRequest.apply)(PostRequest.unapply))
(PostRequest.apply)(PostRequest.unapply))

– unapply - applyは引数を受け取りオブジェクトを返す、unapplyはオブジェクトから引数を返す - つまりapplyで値を適用し、unapplyで値を取り出す、一度applyしているため型安全が保証されると理解した

GETのときの動作
  • 投稿されたメッセージを全て読み出して、htmlテンプレートを呼び出す
POSTのときの動作
  • bindFromRequestを使ってリクエストからパラメータを読み取る
  • foldメソッドを使って結果を処理
    • 第一引数がバリデーション失敗時に呼び出す関数
    • 第二引数がバリデーション成功時に呼び出す関数
@Inject()とは

HTMLテンプレート

@import controllers.Post
@import java.time.format.DateTimeFormatter

@(posts: Seq[Post], form: Form[PostRequest])(implicit messages: Messages)
<!DOCTYPE html>
<html lang="ja">
  <head>
    <title>Scala Text Textboard</title>
    <link rel="stylesheet" href="https://maxcdn.bootstrapcdn.com/bootstrap/3.3.7/css/bootstrap.min.css" integrity="sha384-BVYiiSIFeK1dGmJRAkycuHAHRg32OmUcww7on3RYdg4Va+PmSTsz/K68vbdEjh4u" crossorigin="anonymous">
    <link rel="stylesheet" href="/assets/textboard.css">
  </head>
  <body>
    <div class="container">
      <h1>@Messages("textboard.title")</h1>
      @for(error <- form.errors){
        <p class="text-danger" id="error">@Messages(error.message)</p>
      }
      <form method="POST" action="/" class="form-inline">
        <input type="text" class="form-control" id="post" name="post">
        <button type="submit" class="btn btn-default">@Messages("textboard.send")</button>
      </form>
      <table class="table">
        <thead>
          <tr><th>@Messages("textboard.dateTime")</th><th>@Messages("textboard.message")</th></tr>
        </thead>
        <tbody>
          @for(post <- posts.reverse){
            <tr>
              <td class="post-date">@{
                val formatter = DateTimeFormatter.ofPattern(Messages("textboard.dateFormat"), messages.lang.toLocale)
                post.date.format(formatter)
              }</td>
              <td class="post-body">@post.body</td>
            </tr>
          }
        </tbody>
      </table>
    </div>
  </body>
</html>

  • Twirlは @Messages のようにHTMLの中で @ が付いている場所をScalaのコードだと解釈します。

テンプレートの引数

  • index.scala.html というファイルは views.html.index という関数に変換されます。 - そんで例えばgetメソッド等のロジック内でOk(views.html.index(PostRepository.findAll, form))みたく呼び出される - テンプレートが関数に変換されるあたり関数型の思想なのかなと思った

投稿されたメッセージの表示

          @for(post <- posts.reverse){
            <tr>
              <td class="post-date">@{
                val formatter = DateTimeFormatter.ofPattern(Messages("textboard.dateFormat"), messages.lang.toLocale)
                post.date.format(formatter)
              }</td>
              <td class="post-body">@post.body</td>
            </tr>
          }

@{ } は複数行のScalaのコードを書く場合に便利な記法

その他調べたこと

シールド修飾子とは

コンパニオンオブジェクトの必要性が分からない

コンパニオンオブジェクトって何のためにいるの？ for Rubyist - Qiita

applyって何だっけ？