diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 994aeb6..d88acfb 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -14,7 +14,7 @@ jobs: runs-on: ${{ matrix.os }} env: # The oldest iyi this release supports (README, Requirements). - IYI_VERSION: 0.18.0 + IYI_VERSION: 0.20.0 steps: - uses: actions/checkout@v4 diff --git a/CHANGELOG.md b/CHANGELOG.md index f78cd79..0bcae8b 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,3 +1,52 @@ +## v0.4.0 + +Requires iyi 0.20.0: a `Database` shared by tasks on every worker needs +`spawn_parallel` (0.19.0), the `@[Share]` defs and `SharedProc` that check +the blocks it keeps (0.19.1), and 0.20.0's scheduler. On 0.19.1, tasks that +park on a full pool while the task holding its connections sleeps are +stopped most runs with "deadlock: every fiber is blocked", or wait out +`checkout_timeout`; on 0.20.0 they are not. + +* A `Database` and its `Pool` are `Share`, so one kept in a constant may be + used by parallel tasks on every worker, with no `Share` error and no + warning from the compiler (db#1). v0.3.0's pool kept its bookkeeping with + no lock, and two tasks on different workers were handed one connection: + one read the other's rows, or the runtime panicked with "two fibers + reading one fd". The bookkeeping is behind a `Mutex` now, and a + connection goes in and out of it in a slot that is `Share` on trust. +* A checkout that finds the pool full parks, and its worker runs other + tasks. A release hands its resource, or the room it left, to the checkout + that has waited longest, which a later checkout no longer takes first. +* `Pool#close` and `Database#close` close the idle resources, and each one a + task holds when the task releases it, where they closed it under the task. +* **Breaking:** the blocks a pool or database keeps run on whichever worker + needs them, several at once, so the compiler checks each where it is + written, as it checks a parallel task's: `Pool.new`'s, `Database.new`'s + and `Database.open`'s factory, and `setup_connection`'s block, may capture + only `Share` values. A block that counts in a captured local, or keeps + what it made in a captured array, is refused: count in an `Atomic`, keep + it behind a `Mutex`. +* `Db.open` calls the driver's `ConnectionBuilder#build` from any worker, on + trust: a builder holds what the URI said and only reads it. +* `Pool#stats` reads atomics, and takes no lock. +* A pool knows a resource given back by its `object_id`: a resource is a + class's instance. +* The driver registry is behind a lock, so `Db.open` may run on any worker. +* **Breaking:** `Statement.def_around_query_or_exec`'s hooks run on every + worker, so a hook may capture only `Share` values. A statement learns + there are none, the common case, from an atomic count, without a lock. +* **Breaking:** `Db::Core::LOG` is a `SharedLogger`: std's logger for "db", + which is not `Share`, behind a lock, with its level in an atomic beside + it, so a statement with debug off takes no lock. Its level is read off + std's logger when the first statement asks; set it after that with + `Db::Core::LOG.level =`. A backend is set on std's `Log.for("db")`. +* The dummy driver (`db/testing/dummy_driver`) may be used from every + worker: its counts are atomics, and what it keeps for the tests is behind + a lock. +* `db/parallel_test.iyi` runs db#1's program, twenty times, and the pool's + other cases on twelve workers; `manual/pool_concurrency.iyi` runs its + tasks with `spawn_parallel`. + ## v0.3.0 * `Database#close` and `Pool#close` close every resource. They closed every diff --git a/README.md b/README.md index ee1f812..65736a4 100644 --- a/README.md +++ b/README.md @@ -13,7 +13,10 @@ require: errors are values, types are closed, and tasks are structured. ## Requirements -iyi 0.18.0 or later. +iyi 0.20.0 or later: a `Database` shared by tasks on every worker needs +`spawn_parallel` (0.19.0), the `@[Share]` defs and `SharedProc` that check +the blocks it keeps (0.19.1), and 0.20.0's scheduler, on which a checkout +parked on a full pool is woken when a connection comes back. ## Installation @@ -26,7 +29,7 @@ iyi get github.com/iyilang/db --as db which writes this line to `iyi.mod`: ``` -require github.com/iyilang/db v0.3.0 as db +require github.com/iyilang/db v0.4.0 as db ``` Files then `import db` for the root module, `Db.open` and the driver @@ -89,6 +92,37 @@ The pool is configured from the URI's query string: `initial_pool_size` seconds). `prepared_statements` (true) and `prepared_statements_cache` (true) choose how statements run. +### On every worker + +A `Database` is `Share`, so one opened once and kept in a constant serves +parallel tasks (`spawn_parallel`) on every worker, a web server's handlers +among them, with no `Share` error and no warning from the compiler: + +```iyi +DATABASE = Db.open(ENV["DATABASE_URL"]).or_panic + +def ask(n : Int64) : Int64 + DATABASE.query_one("SELECT $1::int8", n, as: Int64).or_panic +end + +group do |g| + 32.times { |t| g.spawn_parallel { ask(t.to_i64); nil } } +end +``` + +The pool hands each connection to one task at a time, decides which one +under a `Mutex`, and parks a checkout that finds none free - the task, not +its worker - until a release hands it one. The blocks a database keeps run +on whichever worker opens a connection or runs a statement, several at +once, so the compiler checks each where it is written as it checks a +parallel task's: the factory given to `Pool.new`, `Database.new` and +`Database.open`, `setup_connection`'s block and +`Statement.def_around_query_or_exec`'s hooks may capture only `Share` +values. Count in an `Atomic`, keep things behind a `Mutex`. + +`close` closes the idle connections at once, and each one a task holds +when the task releases it. + ### Reading rows into objects ```iyi @@ -115,8 +149,14 @@ property for. `Employee.from_db_row(rs)` reads the current row and ### Logging and instrumentation Every statement logs `Executing query` at debug level to the std/log logger -`"db"` (`Db::Core::LOG`), with the query and its arguments as the entry's -data: `Executing query -- query="select name from contacts where age > ?" args=[18]`. +`"db"`, with the query and its arguments as the entry's data: +`Executing query -- query="select name from contacts where age > ?" args=[18]`. +`Db::Core::LOG` is that logger behind a lock, with its level in an atomic +beside it, so a statement with debug off takes no lock. Its level is read +off std's logger when the first statement asks, and set after that with +`Db::Core::LOG.level = Severity::Debug`; a level given to std's loggers +directly once it has been read is not seen. A backend is set on std's +`Log.for("db")`. `Statement.def_around_query_or_exec { |stmt, args, run| ...; run.call; ... }` wraps every query a driver runs. @@ -125,7 +165,7 @@ wraps every query a driver runs. | module | what it holds | |---|---| | `db` | `Db.open`, `.connect`, `.register_driver`, `.driver_class`, `VERSION` | -| `db/core` | `Driver`, `ConnectionBuilder`, `Database`, `Connection`, `Statement`, `ResultSet`, transactions, `PoolStatement`s, the `QueryMethods`/`SessionMethods`/`StatementMethods`/`BeginTransaction`/`Disposable`/`ConnectionContext` traits, `LOG` | +| `db/core` | `Driver`, `ConnectionBuilder`, `Database`, `Connection`, `Statement`, `ResultSet`, transactions, `PoolStatement`s, the `QueryMethods`/`SessionMethods`/`StatementMethods`/`BeginTransaction`/`Disposable`/`ConnectionContext` traits, `LOG` and its `SharedLogger` | | `db/errors` | `DBError` and its subclasses | | `db/value` | `DBValue`, `CustomValue`, `Any`, `ExecResult` | | `db/pooling` | `Pool(T)`, usable for any resource with a `close` | @@ -152,12 +192,13 @@ crystal-db's. | `DB::MetadataValueConverter.arg_to_log` overloads | `CustomValue#to_log` | R-3 | | `Pool.new` and `Database.new` open the initial connections | `Pool#prefill` and `Database.open(...)` do; the constructors do no IO (a constructor has no error to answer) | III.1 | | `TopLevelTransaction.new` runs `BEGIN` | `Transaction#start` runs `BEGIN`/`SAVEPOINT`; `perform_commit`/`perform_rollback` are what a transaction subclass implements | III.1 | -| `Pool` guards itself with a `Mutex` | no lock: tasks share their group's thread and interleave only where one parks | III.4.5 | +| `Pool` guards itself with a `Mutex` | so does iyi-db's, around its bookkeeping, since tasks run on every worker (`spawn_parallel`); a connection goes in and out of the lock in a slot that is `Share` on trust, as no connection is `Share` | III.4.4, III.4.13 | +| a pool's factory, `setup_connection`'s block and `def_around_query_or_exec`'s hooks may capture anything | they run on whichever worker needs them, so they may capture only `Share` values, checked where they are written (`@[Share]`) | III.4.4: data races are a compile error | | `disconnect!`-style names | no `!` in names | III.1.7 | | `DB::DriverSpecs.run do ... end` (implicit receiver) | `DriverSpecs.run do \|specs\| specs.... end` | no `with ... yield` | | `Connection::Options.from_http_params` read the statement cache default from `prepared_statements` | it reads `prepared_statements_cache` | a crystal-db bug, fixed | -| `Pool#close` closes every other resource, as closing one takes it out of the list being walked | it closes every one | a crystal-db bug, fixed | -| a checkout waiting on a full pool is woken only when a resource goes idle, and each wake-up waits `checkout_timeout` again | it is woken when a resource leaves the pool too, and waits `checkout_timeout` in all | a crystal-db bug, fixed | +| `Pool#close` closes every other resource, as closing one takes it out of the list being walked; and the ones out, under the tasks using them | it closes every idle one, and each one out when it is released | a crystal-db bug, fixed; a task's connection is not closed under it | +| a checkout waiting on a full pool is woken only when a resource goes idle, and each wake-up waits `checkout_timeout` again | a release hands its resource, or the room it left, to the checkout that has waited longest, and a wait lasts `checkout_timeout` in all | a crystal-db bug, fixed | | a closed pool builds a new resource for the next checkout | a checkout of a closed pool, a waiting one too, answers "the pool is closed" | a crystal-db bug, fixed | | the statement cache hands back a statement its user closed | it builds a new one | a crystal-db bug, fixed | @@ -172,6 +213,11 @@ abstract methods say what to implement; the comments on `Driver`, `ConnectionLost.new(connection)`, so the pool retries on another one. Values of the driver's own types are classes inheriting `CustomValue`. +A `Database` that `Db.open` makes calls the driver's `ConnectionBuilder#build` +from whichever worker opens a connection, several at once, and takes that on +trust: a builder holds what the URI said, made once, and `build` only reads +it. + Test a driver with the shared specs: ```iyi @@ -194,11 +240,12 @@ iyi test db # every *_test.iyi ``` The tests are crystal-db's specs, ported; `db/testing/dummy_driver.iyi` is -its dummy driver. Every test also builds from artifacts -(`iyi build --emit-iyimod DIR` then `--use-iyimod DIR`). `manual/` holds the -two programs crystal-db runs by hand: -`iyi run manual/pool_concurrency.iyi -- --duration=3` and -`iyi run manual/load.iyi`. +its dummy driver, safe to use from every worker. `db/parallel_test.iyi` runs +db#1's program and the pool's other cases on twelve workers. Every test also +builds from artifacts (`iyi build --emit-iyimod DIR` then `--use-iyimod +DIR`). `manual/` holds the two programs crystal-db runs by hand: +`iyi run manual/pool_concurrency.iyi -- --duration=3`, whose tasks run on +every worker, and `iyi run manual/load.iyi`. `iyi check` a program, not a module that only queries: a block passed to `Database#query` is typed through the driver's `Statement`, and a module diff --git a/db.iyi b/db.iyi index 89705e9..b0e4323 100644 --- a/db.iyi +++ b/db.iyi @@ -80,24 +80,29 @@ module db # `db/serializable`; a program imports the ones it names. import std/uri::{URI, Params} +import std/mutex::{Mutex} import db/errors::* import db/core::* -pub VERSION = "0.3.0" +pub VERSION = "0.4.0" -DRIVERS = {} of String => Driver.class +# The registered drivers. `Db.open` reads it on whichever worker opens a +# database, so it is behind a lock. +DRIVERS = Mutex.new { {} of String => Driver.class } # The driver class registered for *driver_name*. pub def driver_class(driver_name : String) : Driver.class | DBError - DRIVERS[driver_name]? || - ArgumentError.new(%(no driver was registered for the schema "#{driver_name}", did you maybe forget to require the database driver?)) + found = DRIVERS.get { |drivers| drivers[driver_name]? } + return DBError.new("cancelled while looking up a driver") if found.is_a?(Cancelled) + found || ArgumentError.new(%(no driver was registered for the schema "#{driver_name}", did you maybe forget to require the database driver?)) end # Registers a driver class for a given *driver_name*. Called by driver # implementors, from their module's top level, so importing the driver is # what registers it. pub def register_driver(driver_name : String, driver_class : Driver.class) : Nil - DRIVERS[driver_name] = driver_class + while DRIVERS.lock { |drivers| drivers[driver_name] = driver_class }.is_a?(Cancelled) + end end # Creates a `Database` pool and opens its initial connection(s) as @@ -115,10 +120,29 @@ pub def open(uri : URI | String) : Database | DBError params = Params.parse(parsed.query || "") connection_options = driver.connection_options(params)! pool_options = driver.pool_options(params)! - builder = driver.connection_builder(parsed)! + builder = SharedBuilder.new(driver.connection_builder(parsed)!) Database.open(connection_options, pool_options) { builder.build } end +# A driver's `ConnectionBuilder`, as the factory of the database `open` +# makes holds it: `Share` on trust. The pool calls `build` from whichever +# worker opens a connection, several at once, and a builder holds what the +# URI said and only reads it (`ConnectionBuilder`'s contract). Trusted +# rather than checked because a builder is often not `Share` by its fields +# although nothing writes them: what a driver parsed its URI into is +# written while it is parsed (pg's `ConnInfo`) or holds a `URI` (mysql's). +# `pub` only so that an artifact carries its marker to the programs that +# build `open`. +@[Share] +pub class SharedBuilder + def initialize(@builder : ConnectionBuilder) + end + + def build : Connection | DBError + @builder.build + end +end + # Same as `open`, but the database is yielded and closed when the block is # left. pub def open(uri : URI | String, & : Database -> U) : U | DBError forall U diff --git a/db/core.iyi b/db/core.iyi index c2c2066..320f7fa 100644 --- a/db/core.iyi +++ b/db/core.iyi @@ -27,18 +27,99 @@ module db/core import std/uri::{URI, Params} -import std/log::{Log, Logger} +import std/log::{Log, Logger, Severity} import std/random::{Random} +import std/mutex::{Mutex} +import std/shared_proc::{SharedProc} +import std/list::{List} import std/text import db/errors::* import db/value::* import db/query_params::* import db/pooling::* +# A std/log logger that any worker may write to. std's `Logger` is not +# `Share` - its level and backend are written in place - so it is kept +# behind a lock, and its level is held beside it in an atomic: a statement +# asks the atomic whether it logs, and takes the lock only when it does. +# With debug off, the common case, logging costs every worker one atomic +# read and no lock. +# +# The level is read off std's logger the first time it is asked for - the +# first statement, usually after the program has set up its logging - and +# changes with `level=` after that. A level given to std's loggers +# directly once it has been read (`Log.setup`, `Log.for("db").level =`) is +# not seen here. +pub class SharedLogger + @logger : Mutex(Logger) + # the level's `Severity#value`, or -1 until it is first read. Not named + # `@level`: a setter `level=` marks a field of that name as written, + # whatever the setter does (SPEC.md III.4.4). + @threshold : Atomic(Int32) + + # The logger for *source*, as `Log.for(source)` answers it. + def initialize(source : String) + @threshold = Atomic(Int32).new(-1) + @logger = Mutex.new { Log.for(source) } + end + + # The level below which entries are dropped, as `Logger#level`. + def level : Severity + Severity.from_value?(threshold) || Severity::Info + end + + # Sets the level, as `Logger#level=`: nil hands it back to the parent + # logger's. + def level=(severity : Severity?) : Nil + while true + level = @logger.get { |logger| logger.level = severity; logger.level.value } + unless level.is_a?(Cancelled) + @threshold.set(level) + return + end + end + end + + # Whether an entry at *severity* is written. + def enabled?(severity : Severity) : Bool + severity != Severity::None && severity.value >= threshold + end + + def debug? : Bool + enabled?(Severity::Debug) + end + + # Writes an entry, with the lock held, when the level lets it through. + def log(severity : Severity, message : String, **data) : Nil + return unless enabled?(severity) + @logger.lock { |logger| logger.log(severity, message, **data) } + nil + end + + def debug(message : String, **data) : Nil + log(Severity::Debug, message, **data) + end + + # The level's value, read off std's logger the first time. + private def threshold : Int32 + known = @threshold.get + return known if known >= 0 + while true + level = @logger.get { |logger| logger.level.value } + unless level.is_a?(Cancelled) + # a `level=` that got in first stands + @threshold.compare_and_set(-1, level) + return @threshold.get + end + end + end +end + # The logger every statement writes "Executing query" to, at debug level, # with the query and its arguments as the entry's `query` and `args` data. -# crystal-db's `DB::Log`; its source is "db". -pub LOG = Log.for("db") +# crystal-db's `DB::Log`; its source is "db". Turn the entries on with +# `Db::Core::LOG.level = Severity::Debug`. +pub LOG = SharedLogger.new("db") # A resource with a `close` that runs once: connections, statements, result # sets and transactions. An implementer declares `@closed = false` and a @@ -674,8 +755,13 @@ pub abstract class Statement # puts "#{stmt.command} took #{Time.monotonic - started}" # end # ``` + # + # A hook runs on whichever worker runs the statement, several at once, + # so it is checked where it is written as a parallel task's block is: + # what it captures must be `Share`. + @[Share] def self.def_around_query_or_exec(&hook : Statement, Array(DBValue), Proc(Nil) -> Nil) : Nil - AROUND_HOOKS.add(hook) + AROUND_HOOKS.add(AroundHook.of { |call| hook.call(call[0], call[1], call[2]) }) end # Removes every hook added with `def_around_query_or_exec`. @@ -755,10 +841,10 @@ pub abstract class Statement run.call end - private def wrap_hook(hook : Proc(Statement, Array(DBValue), Proc(Nil), Nil), args : Array(DBValue), inner : Proc(Nil)) : Proc(Nil) + private def wrap_hook(hook : AroundHook, args : Array(DBValue), inner : Proc(Nil)) : Proc(Nil) statement = self.as(Statement) -> do - hook.call(statement, args, inner) + hook.call({statement, args, inner}) nil end end @@ -775,27 +861,47 @@ end impl StatementMethods for Statement end +# A hook `Statement.def_around_query_or_exec` adds, called with the +# statement, its arguments and the proc that runs it. +alias AroundHook = SharedProc(Tuple(Statement, Array(DBValue), Proc(Nil)), Nil) + # The hooks `Statement.def_around_query_or_exec` adds: one list for every # statement class. Module state, not a class variable, which each subclass # would hold its own copy of - a driver's statements would never see the -# hooks added on `Statement`. +# hooks added on `Statement`. Every worker's statements read it, so it is +# `Share`: the list is replaced whole behind a lock when a hook is added, +# and a count beside it lets a statement find there are none, the common +# case, without the lock. class AroundHooks - getter hooks : Array(Proc(Statement, Array(DBValue), Proc(Nil), Nil)) + @hooks : Mutex(List(AroundHook)) + @count : Atomic(Int32) def initialize - @hooks = [] of Proc(Statement, Array(DBValue), Proc(Nil), Nil) + @hooks = Mutex.new { List(AroundHook).new([] of AroundHook) } + @count = Atomic(Int32).new(0) end - def add(hook : Proc(Statement, Array(DBValue), Proc(Nil), Nil)) : Nil - @hooks << hook + def hooks : List(AroundHook) + while true + hooks = @hooks.get { |list| list } + return hooks unless hooks.is_a?(Cancelled) + end + end + + def add(hook : AroundHook) : Nil + while @hooks.update { |list| list.appended(hook) }.is_a?(Cancelled) + end + @count.add(1) end def any? : Bool - !@hooks.empty? + @count.get > 0 end def clear : Nil - @hooks = [] of Proc(Statement, Array(DBValue), Proc(Nil), Nil) + while @hooks.update { |_| List(AroundHook).new([] of AroundHook) }.is_a?(Cancelled) + end + @count.set(0) end end @@ -1068,6 +1174,10 @@ end # A connection factory with a specific configuration. See # `Driver#connection_builder`. +# +# A `Database` that `Db.open` makes calls `build` from whichever worker +# needs a new connection, several at once. A builder holds what the URI +# said, made once by `connection_builder`, and `build` only reads it. pub abstract class ConnectionBuilder # Opens a connection; `ConnectionRefused` when it cannot be established. abstract def build : Connection | DBError @@ -1161,28 +1271,49 @@ end # This can be changed from the `prepared_statements` URI parameter: # # - `prepared_statements` (true, or false, default true) +# +# ## Every worker +# +# A `Database` is `Share`: one kept in a constant may be used from parallel +# tasks (`spawn_parallel`) on every worker. Its pool hands each connection +# to one task at a time (`Pool`), and the blocks it keeps - the connection +# factory and `setup_connection`'s - run on whichever worker opens a +# connection, so each is checked where it is written, as a parallel task's +# block is. pub class Database getter pool : Pool(Connection) @connection_options : Connection::Options - @setup_connection : Proc(Connection, Nil | DBError) + # the block `setup_connection` gave, which every connection the pool + # opens runs, on whichever worker opens it; behind a lock, since a task + # may set it while another opens a connection + @setup : Mutex(SharedProc(Connection, Nil | DBError)?) # Initializes a database with the specified options and connection # factory, for what a URI cannot say, such as a tunnelled connection. No # connection is opened yet: `Database.open` opens the initial ones too. - def initialize(connection_options : Connection::Options, pool_options : Pool::Options, &factory : -> Connection | DBError) - @connection_options = connection_options - @setup_connection = ->(conn : Connection) { nil.as(Nil | DBError) } + # + # The *factory* runs on whichever worker needs a new connection, several + # at once, so what it captures must be `Share`. + @[Share] + def self.new(connection_options : Connection::Options, pool_options : Pool::Options, &factory : -> Connection | DBError) : Database + new(connection_options, pool_options, SharedProc(Nil, Connection | DBError).of { |_| factory.call }) + end + + # A database over a factory already checked: the one `new` with a block + # builds. + def initialize(@connection_options : Connection::Options, pool_options : Pool::Options, factory : SharedProc(Nil, Connection | DBError)) + @setup = Mutex.new { nil.as(SharedProc(Connection, Nil | DBError)?) } @pool = uninitialized Pool(Connection) # in order to use self in the factory proc @pool = Pool(Connection).new(pool_options) do - conn = factory.call + conn = factory.call(nil) next conn if conn.is_a?(DBError) conn.auto_release = false conn.context = self - set_up = @setup_connection.call(conn) - if set_up.is_a?(DBError) + ready = set_up(conn) + if ready.is_a?(DBError) conn.close - next set_up + next ready end conn end @@ -1191,6 +1322,7 @@ pub class Database # A database over the connections *factory* opens, with the initial ones # already open; the error that kept the first of them from opening # otherwise. + @[Share] def self.open(connection_options : Connection::Options, pool_options : Pool::Options, &factory : -> Connection | DBError) : Database | DBError db = Database.new(connection_options, pool_options, &factory) prefilled = db.pool.prefill @@ -1221,13 +1353,19 @@ pub class Database # nil # end! # ``` + # + # The block runs on whichever worker opens a connection, so what it + # captures must be `Share`. + @[Share] def setup_connection(&proc : Connection -> Nil | DBError) : Nil | DBError - @setup_connection = proc + setup = SharedProc(Connection, Nil | DBError).of(&proc) + while @setup.update { |_| setup.as(SharedProc(Connection, Nil | DBError)?) }.is_a?(Cancelled) + end failed = nil.as(DBError?) @pool.each_resource do |conn| # out of the pool for the setup: a statement it runs must not release it conn.auto_release = false - set_up = @setup_connection.call(conn) + set_up = setup.call(conn) if set_up.is_a?(DBError) conn.close failed ||= set_up @@ -1236,7 +1374,15 @@ pub class Database failed end - # Closes all connections to the database. + # What the block `setup_connection` gave answers for *conn*. + private def set_up(conn : Connection) : Nil | DBError + setup = @setup.get { |current| current } + return DBError.new("cancelled while setting up a connection") if setup.is_a?(Cancelled) + setup ? setup.call(conn) : nil + end + + # Closes the database's connections: the idle ones now, and each one a + # task holds when the task releases it. def close : Nil @pool.close end diff --git a/db/database_test.iyi b/db/database_test.iyi index b85f6f6..72ea12a 100644 --- a/db/database_test.iyi +++ b/db/database_test.iyi @@ -13,41 +13,43 @@ end describe "Database" do it "allows connection initialization" do - cnn_setup = [0] + # an atomic, not crystal's array: the setup block runs on whichever + # worker opens a connection, so it captures only `Share` values + cnn_setup = Atomic(Int32).new(0) ::Db.open "dummy://localhost:1027?initial_pool_size=2&max_pool_size=4&max_idle_pool_size=1" do |db| # crystal's constructor opened the initial connections; iyi's pool opens them on prefill db.pool.prefill.or_panic - assert cnn_setup[0] == 0 + assert cnn_setup.get == 0 db.setup_connection do |cnn| - cnn_setup[0] += 1 + cnn_setup.add(1) assert cnn.scalar("a").or_panic == "a" nil end.or_panic - assert cnn_setup[0] == 2 + assert cnn_setup.get == 2 db.using_connection do - assert cnn_setup[0] == 2 + assert cnn_setup.get == 2 db.using_connection do - assert cnn_setup[0] == 2 + assert cnn_setup.get == 2 db.using_connection do - assert cnn_setup[0] == 3 + assert cnn_setup.get == 3 db.using_connection do - assert cnn_setup[0] == 4 + assert cnn_setup.get == 4 end.or_panic # the pool didn't shrink no new initialization should be done next db.using_connection do - assert cnn_setup[0] == 4 + assert cnn_setup.get == 4 end.or_panic end.or_panic # the pool shrink 1. max_idle_pool_size=1 # after the previous end there where 2. db.using_connection do - assert cnn_setup[0] == 4 + assert cnn_setup.get == 4 # so now there will be a new connection created db.using_connection do - assert cnn_setup[0] == 5 + assert cnn_setup.get == 5 end.or_panic end.or_panic end.or_panic diff --git a/db/parallel_test.iyi b/db/parallel_test.iyi new file mode 100644 index 0000000..19dfa2d --- /dev/null +++ b/db/parallel_test.iyi @@ -0,0 +1,192 @@ +# db#1: one `Database` shared by tasks on every worker - the issue's +# program on the dummy driver, a pool too small for its tasks, a close +# while they query, and connections failing and lost under them. +# +# Twelve workers whatever the machine has, as many as db#1 was found +# with: `IyiWorkers.count` is set before the first parallel task. +module db/parallel_test + +import std/time::{Time, Span} +import db +import db/errors::* +import db/core::* +import db/spec::{describe, it, expect_error} +import db/testing/dummy_driver::* + +IyiWorkers.count = 12 + +# The issue's program as it was written: the database a constant, reached +# from the parallel tasks through a function. It compiles with no `Share` +# error and no warning, since `Database` is `Share`. +DATABASE = ::Db.open("dummy://host?max_idle_pool_size=64&initial_pool_size=8").or_panic + +def ask(n : Int64) : Int64 + DATABASE.query_one("?", n, as: Int64).or_panic +end + +# 32 parallel tasks asking 2000 times each for the number they send. +def ask_everything(db : Database) : Nil + group do |g| + 32.times do |t| + g.spawn_parallel do + 2000.times do |k| + n = t.to_i64 * 100000_i64 + k.to_i64 + got = db.query_one("?", n, as: Int64).or_panic + raise "asked #{n}, got #{got}" if got != n + end + nil + end + end + end +end + +# Every connection back in the pool, and none in flight. +def assert_settled(db : Database) : Nil + stats = db.pool.stats + assert stats.in_flight_connections == 0 + assert stats.idle_connections == stats.open_connections, "#{stats.idle_connections} idle of #{stats.open_connections} open" +end + +describe "a Database every worker shares (db#1)" do + it "answers the issue's program, through a constant, with the rows each task asked for" do + group do |g| + 32.times do |t| + g.spawn_parallel do + 2000.times do |k| + n = t.to_i64 * 100000_i64 + k.to_i64 + got = ask(n) + raise "asked #{n}, got #{got}" if got != n + end + nil + end + end + end + assert_settled(DATABASE) + end + + it "answers every task with its own rows, 20 runs over" do + 20.times do + db = ::Db.open("dummy://host?max_idle_pool_size=64&initial_pool_size=8").or_panic + ask_everything(db) + assert_settled(db) + db.close + end + end + + # More checkouts wait than there are workers: had a waiting checkout + # held its worker, the task holding the connections would never wake + # from its sleep to give them back, and every wait would time out. + it "parks the checkouts a full pool cannot serve, not their workers, and resumes them" do + db = ::Db.open("dummy://host?initial_pool_size=1&max_pool_size=2&max_idle_pool_size=2&checkout_timeout=10").or_panic + go = Channel(Nil).new + served = Atomic(Int32).new(0) + most = Atomic(Int32).new(0) + started = Time.monotonic + group do |g| + g.spawn_parallel do + a = db.checkout.or_panic + b = db.checkout.or_panic + go.close + sleep 300 + a.release + b.release + nil + end + 40.times do |t| + g.spawn_parallel do + go.receive + n = t.to_i64 + got = db.query_one("?", n, as: Int64).or_panic + raise "asked #{n}, got #{got}" if got != n + open = db.pool.stats.open_connections + seen = most.get + while open > seen && !most.compare_and_set(seen, open)[1] + seen = most.get + end + served.add(1) + nil + end + end + end + waited = Time.monotonic - started + assert served.get == 40 + assert most.get <= 2, "#{most.get} connections open, of at most 2" + assert waited >= Span.milliseconds(300) + assert waited < Span.seconds(5), "took #{waited}" + assert_settled(db) + db.close + end + + it "closes while tasks on every worker are querying" do + DummyConnection.clear_connections + db = ::Db.open("dummy://host?initial_pool_size=4&max_pool_size=8&max_idle_pool_size=8").or_panic + answered = Atomic(Int32).new(0) + refused = Atomic(Int32).new(0) + group do |g| + 32.times do |t| + g.spawn_parallel do + k = 0 + while true + n = t.to_i64 * 100000_i64 + k.to_i64 + got = db.query_one("?", n, as: Int64) + if got.is_a?(DBError) + raise "expected the pool closed, got #{got.message}" if got.message != "the pool is closed" + refused.add(1) + break + end + raise "asked #{n}, got #{got}" if got != n + answered.add(1) + k += 1 + end + nil + end + end + sleep 100 + db.close + end + assert refused.get == 32 + assert answered.get > 0 + # the connections the tasks held were closed as they were released + assert db.pool.stats.open_connections == 0 + assert DummyConnection.connections.all?(&.closed?) + end + + it "takes back the connection a statement failed on, and drops a lost one" do + DummyConnection.clear_connections + db = ::Db.open("dummy://host?initial_pool_size=4&max_pool_size=8&max_idle_pool_size=8&retry_attempts=2&retry_delay=0").or_panic + group do |g| + 32.times do |t| + g.spawn_parallel do + 200.times do |k| + n = t.to_i64 * 100000_i64 + k.to_i64 + case k % 4 + when 0 + # the statement fails; its connection goes back to the pool + expect_error(db.exec("raise"), DBError, "forced exception due to query") + when 1 + # the connection is lost under the task that holds it, which + # gives it back closed: the pool drops it + db.using_connection do |cnn| + cnn.as(DummyConnection).disconnect + expect_error(cnn.exec("x"), ConnectionLost) + nil + end.or_panic + when 2 + # every connection the statement is tried on is lost + expect_error(db.exec("raise ConnectionLost"), PoolRetryAttemptsExceeded) + else + got = db.query_one("?", n, as: Int64).or_panic + raise "asked #{n}, got #{got}" if got != n + end + end + nil + end + end + end + assert_settled(db) + open = DummyConnection.connections.reject(&.closed?) + assert db.pool.stats.open_connections == open.size + assert open.all? { |cnn| db.pool.is_available?(cnn) } + db.close + end +end diff --git a/db/pool_test.iyi b/db/pool_test.iyi index 30bc7fd..b619112 100644 --- a/db/pool_test.iyi +++ b/db/pool_test.iyi @@ -1,7 +1,14 @@ # crystal-db's spec/pool_spec.cr. +# +# A factory runs on whichever worker builds a resource now, so it may +# capture only `Share` values: where crystal's examples counted builds in a +# captured integer or kept the resources in a captured array, these count +# in an `Atomic` and keep them in a `Built`. module db/pool_test import std/time::{Time, Span} +import std/mutex::{Mutex} +import std/list::{List} import db/errors::* import db/core::{Disposable} import db/pooling::{Pool} @@ -76,8 +83,29 @@ end impl Disposable for Closable end +# The resources a factory built, oldest first. +class Built + @made : Mutex(Array(Made(Closable))) + + def initialize + @made = Mutex.new { [] of Made(Closable) } + end + + def make : Closable + resource = Closable.new + made = Made.new(resource) + @made.lock { |all| all << made }.or_panic + resource + end + + def all : Array(Closable) + @made.get { |all| List(Made(Closable)).new(all) }.or_panic.map(&.value) + end +end + # Crystal's `Pool.new` opened the `initial_pool_size` resources itself; iyi's # `prefill` does it. +@[Share] def create_pool(options : Pool::Options = Pool::Options.new, &factory : -> Closable) : Pool(Closable) pool = Pool(Closable).new(options) { factory.call } pool.prefill.or_panic @@ -86,9 +114,9 @@ end describe "Db::Pool" do it "should use proc to create objects" do - block_called = 0 - pool = create_pool(Pool::Options.new(initial_pool_size: 3)) { block_called += 1; Closable.new } - assert block_called == 3 + block_called = Atomic(Int32).new(0) + pool = create_pool(Pool::Options.new(initial_pool_size: 3)) { block_called.add(1); Closable.new } + assert block_called.get == 3 end it "should get resource" do @@ -99,10 +127,10 @@ describe "Db::Pool" do end it "should be available if not checkedout" do - resource : Closable? = nil - pool = create_pool(Pool::Options.new(initial_pool_size: 1)) { c = Closable.new; resource = c; c } - r = resource - assert !r.nil? && pool.is_available?(r) + built = Built.new + pool = create_pool(Pool::Options.new(initial_pool_size: 1)) { built.make } + r = built.all[0] + assert pool.is_available?(r) end it "should not be available if checkedout" do @@ -152,40 +180,46 @@ describe "Db::Pool" do end it "should create new if max was not reached" do - block_called = 0 - pool = create_pool(Pool::Options.new(max_pool_size: 2, initial_pool_size: 1)) { block_called += 1; Closable.new } - assert block_called == 1 + block_called = Atomic(Int32).new(0) + pool = create_pool(Pool::Options.new(max_pool_size: 2, initial_pool_size: 1)) { block_called.add(1); Closable.new } + assert block_called.get == 1 pool.checkout.or_panic - assert block_called == 1 + assert block_called.get == 1 pool.checkout.or_panic - assert block_called == 2 + assert block_called.get == 2 end it "should reuse returned resources" do - all = [] of Closable - pool = create_pool(Pool::Options.new(max_pool_size: 2, initial_pool_size: 1)) { c = Closable.new; all << c; c } + built = Built.new + pool = create_pool(Pool::Options.new(max_pool_size: 2, initial_pool_size: 1)) { built.make } pool.checkout.or_panic b1 = pool.checkout.or_panic pool.release b1 b2 = pool.checkout.or_panic assert b1 == b2 - assert all.size == 2 + assert built.all.size == 2 end - it "should close available and total" do - all = [] of Closable - pool = create_pool(Pool::Options.new(max_pool_size: 2, initial_pool_size: 1)) { c = Closable.new; all << c; c } + # crystal's `close` closed the resources out too, under the task using + # them; a resource out is closed when it is released. + it "should close available, and one out when it is released" do + built = Built.new + pool = create_pool(Pool::Options.new(max_pool_size: 2, initial_pool_size: 1)) { built.make } a = pool.checkout.or_panic b = pool.checkout.or_panic pool.release b + all = built.all assert all.size == 2 assert !all[0].closed? assert !all[1].closed? pool.close - assert all[0].closed? - assert all[1].closed? + assert !a.closed? + assert b.closed? + pool.release a + assert a.closed? + assert pool.stats.open_connections == 0 end it "should timeout" do @@ -202,12 +236,13 @@ describe "Db::Pool" do end it "should close if max idle amount is reached" do - all = [] of Closable - pool = create_pool(Pool::Options.new(max_pool_size: 3, max_idle_pool_size: 1)) { c = Closable.new; all << c; c } + built = Built.new + pool = create_pool(Pool::Options.new(max_pool_size: 3, max_idle_pool_size: 1)) { built.make } pool.checkout.or_panic pool.checkout.or_panic pool.checkout.or_panic + all = built.all assert all.size == 3 assert !all.any?(&.closed?) pool.release all[0] @@ -238,17 +273,18 @@ describe "Db::Pool" do end it "should create resource after max_pool was reached if idle forced some close up" do - all = [] of Closable - pool = create_pool(Pool::Options.new(max_pool_size: 3, max_idle_pool_size: 1)) { c = Closable.new; all << c; c } + built = Built.new + pool = create_pool(Pool::Options.new(max_pool_size: 3, max_idle_pool_size: 1)) { built.make } pool.checkout.or_panic pool.checkout.or_panic pool.checkout.or_panic + all = built.all pool.release all[0] pool.release all[1] pool.checkout.or_panic pool.checkout.or_panic - assert all.size == 4 + assert built.all.size == 4 end it "counts no resource in flight after the factory panicked" do @@ -272,24 +308,35 @@ describe "Db::Pool" do end end - it "waits checkout_timeout in all, however often it is woken" do - pool = create_pool(Pool::Options.new(max_pool_size: 1, checkout_timeout: 0.3)) { Closable.new } + # A release hands its resource to the checkout that has waited longest, + # and a checkout made after the release does not take it first: crystal's + # pool put it back idle and woke the waiter, which found it gone. + it "hands a released resource to the waiting checkout" do + pool = create_pool(Pool::Options.new(max_pool_size: 1, checkout_timeout: 1.0)) { Closable.new } a = pool.checkout.or_panic - waited = [] of Span group do |g| g.spawn do - started = Time.monotonic - expect_error(pool.checkout, PoolTimeout) - waited << Time.monotonic - started + got = pool.checkout.or_panic + assert got == a + sleep 50 + pool.release got 0 end - sleep 200 - # the release wakes the waiter, and the resource is taken again before - # the waiter runs + sleep 100 pool.release a - pool.checkout.or_panic + # the waiter has it: this checkout waits until the waiter releases it + assert pool.checkout.or_panic == a end - assert waited[0] < Span.milliseconds(450) + end + + it "waits checkout_timeout in all" do + pool = create_pool(Pool::Options.new(max_pool_size: 1, checkout_timeout: 0.3)) { Closable.new } + pool.checkout.or_panic + started = Time.monotonic + expect_error(pool.checkout, PoolTimeout) + waited = Time.monotonic - started + assert waited >= Span.milliseconds(290) + assert waited < Span.milliseconds(600) end it "takes a checkout_timeout too long for a timer as waiting" do @@ -323,10 +370,10 @@ describe "Db::Pool" do end it "builds nothing once it is closed" do - built = 0 - pool = create_pool { built += 1; Closable.new } + built = Atomic(Int32).new(0) + pool = create_pool { built.add(1); Closable.new } pool.close expect_error(pool.checkout, DBError, "the pool is closed") - assert built == 1 + assert built.get == 1 end end diff --git a/db/pooling.iyi b/db/pooling.iyi index 1f0ce92..3930b4e 100644 --- a/db/pooling.iyi +++ b/db/pooling.iyi @@ -3,22 +3,55 @@ # # The resource type is duck-typed as it is in crystal-db: `close` is # required; `closed?`, `before_checkout` and `after_release` are called when -# the resource has them. +# the resource has them. A resource is an object, a class's instance: the +# pool knows the one it is given back by its `object_id`. # -# crystal-db guards the pool with a `Mutex` for multi-threaded runtimes. iyi's -# tasks share their group's thread and interleave only where one parks -# (SPEC.md III.4.5), so the bookkeeping between two parks is already atomic and -# the pool holds no lock. `@inflight` keeps counting the resources being built -# while their builder is parked, as it does in crystal-db, and `each_resource` -# takes each resource out of the pool while its block may park. +# **Every worker may share one pool.** Since iyi 0.19.0 a program's tasks +# run on a pool of worker threads (`spawn_parallel`), and a pool kept in a +# constant is reached by tasks on all of them at once. v0.3.0 kept its +# bookkeeping with no lock, from when iyi's tasks shared one thread, and two +# workers checking out at once were handed one connection (db#1). Now: +# +# * Which resource is idle, which is out and how many are being built is +# decided behind one `Mutex` (std/mutex), in a `PoolBook`. A task that +# finds the lock held parks, and its worker runs other tasks meanwhile. +# * A resource goes into the book and out of it inside a `PoolSlot`, the one +# type `Share` is trusted for here. A connection is not `Share`, being a +# socket and buffers that one task uses at a time, so nothing else lets +# one cross the lock; the book hands each slot to one task at a time. +# * Opening and closing resources, and their hooks, run outside the lock, in +# the task that holds the resource, so a slow connect does not hold up +# every checkout. +# * A checkout that finds no resource free and no room for one parks on a +# channel of its own, registered in the book under the same lock that +# found nothing, so no release can slip in between. A release hands its +# resource straight to the checkout that has waited longest, which no +# other checkout can then take first. +# * The factory runs on whichever worker builds a resource, several at once. +# `Pool.new` is `@[Share]`, so the compiler checks its block where it is +# written, as it checks a parallel task's, and the pool keeps it as a +# `SharedProc`. +# * The counts `stats` answers are atomics, written with the lock held and +# read without it. +# +# So `Pool(T)` is `Share` by its fields, whatever `T` is, and a parallel task +# may hold one. It says so as well, `@[Share(owns: true)]`: a program that +# builds from db's artifacts reads the marker rather than the fields, and a +# generic type found `Share` by its fields is exported as one that is `Share` +# when its arguments are - which `Pool(Connection)` is not, a connection not +# being `Share`. The pool owns its resources as `Mutex(T)` owns its value: +# a task reaches one only between a checkout and its release. module db/pooling import std/uri::{Params} import std/time::{Time} import std/annotations::{Deprecated} +import std/mutex::{Mutex} +import std/shared_proc::{SharedProc} import db/errors::* import db/query_params::* +@[Share(owns: true)] pub class Pool(T) # How a pool is sized and how long it waits. Written out rather than with # `record`, whose struct is not `pub` and so not part of an artifact. @@ -88,25 +121,42 @@ pub class Pool(T) # Pool state - # every open resource managed by this pool - @total : Array(T) - # resources available for checkout - @idle : Set(T) - # resources being built (in neither *@idle* nor *@total* yet) - @inflight : Int32 - # set by `close`: a closed pool builds and hands out nothing - @closed = false - - # tells a waiting checkout that a resource was released, or left the pool - @availability_channel : Channel(Nil) - - @factory : Proc(T | DBError) + # every resource, idle or out, and the checkouts waiting for one + @book : Mutex(PoolBook(T)) + @factory : SharedProc(Nil, T | DBError) + # what `stats` reads without the lock, written with it held after every + # change to the book + @open_count : Atomic(Int32) + @idle_count : Atomic(Int32) + @inflight_count : Atomic(Int32) # Builds the pool without opening anything: `prefill` opens the # `initial_pool_size` resources, and a checkout opens one when none is idle. # Crystal's constructor opened them itself; here opening can fail, and a # constructor has no error to answer. - def initialize(pool_options : Options = Options.new, &factory : -> T | DBError) + # + # The *factory* runs on whichever worker needs a new resource, several at + # once, so it is checked where it is written as a parallel task's block + # is: what it captures must be `Share`. + @[Share] + def self.new(pool_options : Options = Options.new, &factory : -> T | DBError) : Pool(T) + new(pool_options, SharedProc(Nil, T | DBError).of { |_| factory.call }) + end + + @[Deprecated("Use `#new` with Pool::Options instead")] + @[Share] + def self.new(initial_pool_size : Int32 = 1, max_pool_size : Int32 = 0, max_idle_pool_size : Int32 = 1, checkout_timeout : Float64 = 5.0, + retry_attempts : Int32 = 1, retry_delay : Float64 = 0.2, &factory : -> T | DBError) : Pool(T) + options = Options.new( + initial_pool_size: initial_pool_size, max_pool_size: max_pool_size, + max_idle_pool_size: max_idle_pool_size, checkout_timeout: checkout_timeout, + retry_attempts: retry_attempts, retry_delay: retry_delay) + new(options, SharedProc(Nil, T | DBError).of { |_| factory.call }) + end + + # A pool over a factory already checked: the one `new` with a block + # builds. + def initialize(pool_options : Options, @factory : SharedProc(Nil, T | DBError)) @initial_pool_size = pool_options.initial_pool_size @max_pool_size = pool_options.max_pool_size @max_idle_pool_size = pool_options.max_idle_pool_size @@ -114,52 +164,46 @@ pub class Pool(T) @retry_attempts = pool_options.retry_attempts @retry_delay = pool_options.retry_delay - @total = [] of T - @idle = Set(T).new - @inflight = 0 - @availability_channel = Channel(Nil).new - @factory = -> { factory.call.as(T | DBError) } - end - - @[Deprecated("Use `#new` with Pool::Options instead")] - def initialize(initial_pool_size : Int32 = 1, max_pool_size : Int32 = 0, max_idle_pool_size : Int32 = 1, checkout_timeout : Float64 = 5.0, - retry_attempts : Int32 = 1, retry_delay : Float64 = 0.2, &factory : -> T | DBError) - initialize( - Options.new( - initial_pool_size: initial_pool_size, max_pool_size: max_pool_size, - max_idle_pool_size: max_idle_pool_size, checkout_timeout: checkout_timeout, - retry_attempts: retry_attempts, retry_delay: retry_delay), - &factory) + @book = Mutex.new { PoolBook(T).new } + @open_count = Atomic(Int32).new(0) + @idle_count = Atomic(Int32).new(0) + @inflight_count = Atomic(Int32).new(0) end # Opens resources until `initial_pool_size` of them are open. The first # resource the factory cannot build is answered and the rest are not # attempted. def prefill : Nil | DBError - while @total.size < @initial_pool_size - build_resource! + count = @initial_pool_size + while true + reserved = @book.get { |book| room = book.reserve(count); note(book); room } + next if reserved.is_a?(Cancelled) + break unless reserved + build(idle: true)! end nil end - # Closes every resource of the pool, and answers every checkout from now - # on, the waiting ones too, with an error. + # Closes the pool: every checkout from now on, the waiting ones too, is + # answered "the pool is closed". The idle resources are closed now; one a + # task holds is closed when the task releases it, so `close` is safe + # while other tasks are using the pool. def close : Nil - @closed = true - # closing a connection deletes it from `@total`: closed from a copy - resources = @total - @total = [] of T - @idle = Set(T).new - resources.each &.close - while notify_available + while @book.lock { |book| book.close; note(book) }.is_a?(Cancelled) + end + while true + slot = @book.get { |book| idle = book.close_idle; note(book); idle } + next if slot.is_a?(Cancelled) + break unless slot + slot.resource.close end end def stats : Stats Stats.new( - open_connections: @total.size, - idle_connections: @idle.size, - in_flight_connections: @inflight, + open_connections: @open_count.get, + idle_connections: @idle_count.get, + in_flight_connections: @inflight_count.get, max_connections: @max_pool_size, ) end @@ -167,27 +211,7 @@ pub class Pool(T) # An idle resource, a new one when the pool may grow, or one released # within `checkout_timeout`; `PoolTimeout` when none came. def checkout : T | DBError - deadline = nil.as(Float64?) - resource = nil - until resource - return DBError.new("the pool is closed") if @closed - if @idle.empty? - if can_increase_pool? - resource = build_resource! - else - # one deadline for every wait: a woken checkout that finds the - # resource taken waits only what is left of it - limit = deadline || now_ms + @checkout_timeout * 1000 - deadline = limit - wait_for_available(limit - now_ms)! - resource = pick_available - end - else - resource = pick_available - end - end - - @idle.delete resource + resource = take! if resource.responds_to?(:before_checkout) resource.before_checkout end @@ -201,25 +225,16 @@ pub class Pool(T) yield resource end + # Gives back a resource `checkout` answered: to the checkout that has + # waited longest, else idle, else - more idle than `max_idle_pool_size`, + # or the pool closed - it is closed. A closed resource just leaves the + # pool. def release(resource : T) : Nil - # `close` closed every resource the pool had - return if @closed - - if resource.responds_to?(:closed?) && resource.closed? - @total.delete(resource) - elsif can_increase_idle_pool - @idle.add resource - if resource.responds_to?(:after_release) - resource.after_release - end - else - resource.close - @total.delete(resource) + closed = resource.responds_to?(:closed?) && resource.closed? + if !closed && resource.responds_to?(:after_release) + resource.after_release end - - # a waiting checkout takes the idle resource, or builds one in the room - # this one left - notify_available + give_back(resource, closed, @max_idle_pool_size) end # Runs the block again while it answers a lost or refused resource. Every @@ -227,10 +242,14 @@ pub class Pool(T) # grow); after those, each attempt waits `retry_delay` seconds first. # `PoolRetryAttemptsExceeded` once `retry_attempts` more are spent. def retry(& : -> U) : U | DBError forall U - current_available = @idle.size + # read off the counts, without the lock: the number of quick attempts + # need not be exact + current_available = @idle_count.get # if the pool hasn't reached the max size, allow 1 attempt # to make a new resource if needed without sleeping - current_available += 1 if can_increase_pool? + if @max_pool_size == 0 || @open_count.get + @inflight_count.get < @max_pool_size + current_available += 1 + end (current_available + @retry_attempts).times do |i| if i >= current_available @@ -252,88 +271,141 @@ pub class Pool(T) PoolRetryAttemptsExceeded.new end - # Yields every idle resource. Each is out of the pool while its block - # runs, so a checkout cannot take it while the block parks, and goes back - # unless the block closed it. + # Yields every idle resource. They are all out of the pool while the + # blocks run, so no checkout takes one meanwhile, and each goes back + # unless its block closed it - idle again, as many as there were, even + # past `max_idle_pool_size`. def each_resource(& : T -> Nil) : Nil - @idle.to_a.each do |resource| - # an earlier block parked, and a checkout took this one meanwhile - next unless @idle.includes?(resource) - @idle.delete resource + taken = [] of T + while true + slot = @book.get { |book| idle = book.take_idle; note(book); idle } + next if slot.is_a?(Cancelled) + break unless slot + taken << slot.resource + end + taken.each do |resource| yield resource - if @closed || (resource.responds_to?(:closed?) && resource.closed?) - @total.delete(resource) - else - @idle.add resource - end - notify_available + give_back(resource, resource.responds_to?(:closed?) && resource.closed?, 2147483647) end end def is_available?(resource : T) : Bool - @idle.includes?(resource) + id = resource.object_id + while true + idle = @book.get { |book| book.idle?(id) } + return idle unless idle.is_a?(Cancelled) + end end + # Takes a resource out of the pool for good: it was lost, or closed. def delete(resource : T) : Nil - @total.delete(resource) - @idle.delete(resource) - notify_available - end - - private def build_resource : T | DBError - @inflight += 1 - defer @inflight -= 1 - resource = @factory.call - return resource if resource.is_a?(DBError) - if @closed - # built while the pool was being closed - resource.close - return DBError.new("the pool is closed") + id = resource.object_id + while @book.lock { |book| book.delete(id); note(book) }.is_a?(Cancelled) end - @total << resource - @idle.add resource - resource end - private def can_increase_pool? : Bool - @max_pool_size == 0 || @total.size + @inflight < @max_pool_size + # The resource for a checkout: an idle one, a new one when there is room, + # or one handed over by a release. + private def take : T | DBError + max = @max_pool_size + step = @book.get { |book| taken = book.take(max, nil); note(book); taken } + return DBError.new("the checkout was cancelled") if step.is_a?(Cancelled) + if step == PoolStep::Wait + # Nothing free: wait on a channel of its own, registered by the same + # look that finds nothing free, so no release comes in between. + waiter = Channel(PoolSlot(T) | PoolStep).new(1) + step = @book.get { |book| taken = book.take(max, waiter); note(book); taken } + return DBError.new("the checkout was cancelled") if step.is_a?(Cancelled) + step = wait(waiter)! if step == PoolStep::Wait + end + case step + when PoolSlot(T) + step.resource + when PoolStep::Build + build(idle: false) + else + DBError.new("the pool is closed") + end end - private def can_increase_idle_pool : Bool - @idle.size < @max_idle_pool_size + # Parks until a release hands *waiter* a resource, room to build one or + # the news that the pool closed, or until `checkout_timeout` is up. + private def wait(waiter : Channel(PoolSlot(T) | PoolStep)) : PoolSlot(T) | PoolStep | DBError + sent = nil.as(PoolSlot(T) | PoolStep | Nil) + select + when got = waiter.receive + sent = got if got.is_a?(PoolSlot(T)) || got.is_a?(PoolStep) + when timeout(timeout_ms(@checkout_timeout * 1000)) + end + return sent if sent + # Timed out, or cancelled: off the list, unless a release got to this + # checkout first, and what it sent, under the lock, is in the channel. + removed = false + while true + got = @book.get { |book| book.unwait(waiter) } + unless got.is_a?(Cancelled) + removed = got + break + end + end + return PoolTimeout.new("Could not check out a connection in #{@checkout_timeout} seconds") if removed + left = waiter.take_now + left.is_a?(ChannelClosed) ? DBError.new("the pool is closed") : left end - private def pick_available : T? - @idle.each { |resource| return resource } - nil + # Builds a resource the book counted in flight: the checkout's own + # (*idle* false), or an idle one (`prefill`). + private def build(idle : Bool) : T | DBError + settled = false + # a factory that panics leaves no resource counted in flight + defer (build_failed unless settled) + resource = @factory.call(nil) + if resource.is_a?(DBError) + settled = true + build_failed + return resource + end + slot = PoolSlot(T).hold(resource) + kept = false + while true + got = @book.get { |book| open = book.built(slot, idle); note(book); open } + unless got.is_a?(Cancelled) + kept = got + break + end + end + settled = true + unless kept + # built while the pool was being closed + resource.close + return DBError.new("the pool is closed") + end + resource end - # Wakes one waiting checkout, if there is one. - private def notify_available : Bool - woken = false - select - when @availability_channel.send(nil) - woken = true - else + private def build_failed : Nil + while @book.lock { |book| book.build_failed; note(book) }.is_a?(Cancelled) end - woken end - private def wait_for_available(ms : Float64) : Nil | DBError - available = false - if ms > 0 - select - when @availability_channel.receive - available = true - when timeout(timeout_ms(ms)) + private def give_back(resource : T, closed : Bool, max_idle : Int32) : Nil + id = resource.object_id + outcome = PoolReturn::Gone + while true + got = @book.get { |book| given = book.give_back(id, closed, max_idle); note(book); given } + unless got.is_a?(Cancelled) + outcome = got + break end end - return nil if available - PoolTimeout.new("Could not check out a connection in #{@checkout_timeout} seconds") + resource.close if outcome == PoolReturn::Close end - private def now_ms : Float64 - Time.monotonic.total_milliseconds + # The counts `stats` reads, from the book; called with the lock held. + private def note(book : PoolBook(T)) : Nil + @open_count.set(book.open_size) + @idle_count.set(book.idle_size) + @inflight_count.set(book.inflight) end # *ms* as a timer's milliseconds: rounded up, and a wait too long for an @@ -344,3 +416,200 @@ pub class Pool(T) ms.ceil.to_i32 end end + +# A resource as the book holds it, and the one claim of `Share` in this +# module (`owns: true`, as `Mutex(T)` makes it): the book hands a slot to +# one task at a time - a checkout takes it out of the lock, and only a +# release puts it back - and only the task holding the slot reaches its +# resource, so the resource is never in two tasks' hands although it need +# not be `Share` itself. Made by `Pool` alone (`hold`), and `pub` only so +# that an artifact carries its marker to the programs that instantiate a +# pool. +@[Share(owns: true)] +pub class PoolSlot(T) + getter resource : T + getter id : UInt64 + + protected def self.hold(resource : T) : PoolSlot(T) + new(resource) + end + + private def initialize(@resource : T) + @id = @resource.object_id + end +end + +# What the book tells a checkout to do, when it has no resource to give it. +enum PoolStep + # Room for one more: build it. The book counts it in flight already. + Build + # Nothing free and no room: wait for a release. + Wait + # The pool is closed. + Closed +end + +# What became of a resource given back to the book. +enum PoolReturn + # Idle again, or handed to a waiting checkout. + Kept + # Out of the pool, and the task that gave it back closes it: no room for + # it idle, or the pool is closed. + Close + # Out of the pool already: closed, or not this pool's. + Gone +end + +# A pool's bookkeeping, only ever reached behind the pool's `Mutex`. Its +# methods decide and return; nothing here opens, closes or waits. +class PoolBook(T) + # every open resource, idle or out, by object id + @open : Hash(UInt64, PoolSlot(T)) + # the idle ones, the longest idle first + @idle : Array(PoolSlot(T)) + # the checkouts parked for a resource, the longest waiting first; each + # channel holds one value, so a send to it never parks + @waiters : Array(Channel(PoolSlot(T) | PoolStep)) + # resources being built, in neither list yet + getter inflight : Int32 = 0 + @closed = false + + def initialize + @open = {} of UInt64 => PoolSlot(T) + @idle = [] of PoolSlot(T) + @waiters = [] of Channel(PoolSlot(T) | PoolStep) + end + + def open_size : Int32 + @open.size + end + + def idle_size : Int32 + @idle.size + end + + def idle?(id : UInt64) : Bool + @idle.any? { |slot| slot.id == id } + end + + # An idle slot, room to build, or a wait - registered when *waiter* is + # given, so a release after this finds it. + def take(max : Int32, waiter : Channel(PoolSlot(T) | PoolStep)?) : PoolSlot(T) | PoolStep + return PoolStep::Closed if @closed + if slot = @idle.shift? + return slot + end + if room?(max) + @inflight += 1 + return PoolStep::Build + end + @waiters << waiter if waiter + PoolStep::Wait + end + + # *waiter* off the list, its wait over: false when a release had sent it + # something already. + def unwait(waiter : Channel(PoolSlot(T) | PoolStep)) : Bool + !@waiters.delete(waiter).nil? + end + + # Room for one more resource, open or being built. + def room?(max : Int32) : Bool + max == 0 || @open.size + @inflight < max + end + + # One more build for `prefill`, while fewer than *count* are open. + def reserve(count : Int32) : Bool + return false if @closed || @open.size + @inflight >= count + @inflight += 1 + true + end + + # A reserved build, done: the resource is the builder's own, or idle + # (*idle*), or the longest waiting checkout's. False when the pool closed + # meanwhile, and the builder closes it. + def built(slot : PoolSlot(T), idle : Bool) : Bool + @inflight -= 1 + return false if @closed + @open[slot.id] = slot + if idle + if waiter = @waiters.shift? + waiter.put_now(slot) + else + @idle << slot + end + end + true + end + + # A reserved build that failed: its room is free again. + def build_failed : Nil + @inflight -= 1 + pass_room + end + + # The resource *id*, given back by the task that held it. + def give_back(id : UInt64, closed : Bool, max_idle : Int32) : PoolReturn + slot = @open[id]? + # not this pool's, or deleted already, or given back twice + return PoolReturn::Gone if slot.nil? || @idle.includes?(slot) + if closed + @open.delete(id) + pass_room + return PoolReturn::Gone + end + if @closed + @open.delete(id) + return PoolReturn::Close + end + if waiter = @waiters.shift? + waiter.put_now(slot) + return PoolReturn::Kept + end + if @idle.size < max_idle + @idle << slot + return PoolReturn::Kept + end + @open.delete(id) + PoolReturn::Close + end + + # The resource *id* out for good, idle or not. + def delete(id : UInt64) : Nil + slot = @open.delete(id) + return unless slot + @idle.delete(slot) + pass_room + end + + # An idle slot taken out, to be yielded by `each_resource`. + def take_idle : PoolSlot(T)? + @idle.shift? + end + + # An idle slot out of the pool, for `Pool#close` to close. + def close_idle : PoolSlot(T)? + slot = @idle.shift? + @open.delete(slot.id) if slot + slot + end + + # Every waiting checkout is told the pool closed; resources out are + # closed as they come back. + def close : Nil + @closed = true + while waiter = @waiters.shift? + waiter.put_now(PoolStep::Closed) + end + end + + # Room for one more came free: the checkout that has waited longest + # builds in it. + private def pass_room : Nil + return if @closed + if waiter = @waiters.shift? + @inflight += 1 + waiter.put_now(PoolStep::Build) + end + end +end diff --git a/db/statement_test.iyi b/db/statement_test.iyi index a3ad533..30d45d1 100644 --- a/db/statement_test.iyi +++ b/db/statement_test.iyi @@ -2,6 +2,7 @@ module db/statement_test import std/log::{Log, Entry, MemoryBackend, Severity} +import std/mutex::{Mutex} import db import db/errors::* import db/core::* @@ -12,13 +13,15 @@ def dummy_prepared(cnn : DummyConnection, query : String) : DummyStatement cnn.prepared(query).or_panic.as(DummyStatement) end -# The "Executing query" entries LOG wrote while running the block. +# The "Executing query" entries LOG wrote while running the block. The +# backend is set on std's logger for "db", the one `LOG` keeps behind its +# lock: a `MemoryBackend` is not `Share`, so no lock's block may take it. def capture_log(& : -> Nil) : Array(Entry) backend = MemoryBackend.new - LOG.backend = backend + Log.for("db").backend = backend previous_level = LOG.level LOG.level = Severity::Debug - defer LOG.backend = nil + defer Log.for("db").backend = nil defer LOG.level = previous_level yield backend.entries.select { |e| e.severity == Severity::Debug && e.message == "Executing query" } @@ -364,16 +367,19 @@ describe "Statement" do # has no open classes, so the hooks are a registry on `Statement`. describe "def_around_query_or_exec" do it "runs every hook around the driver's statements, the first added outermost" do - seen = [] of String + # a hook runs on whichever worker runs the statement: what it keeps + # goes behind a lock, and what it captures is `Share` + seen = Mutex.new { [] of String } Statement.def_around_query_or_exec do |stmt, args, run| - seen << "outer before #{stmt.command} #{args.size}" + before = "outer before #{stmt.command} #{args.size}" + seen.lock { |lines| lines << before }.or_panic run.call - seen << "outer after" + seen.lock { |lines| lines << "outer after" }.or_panic end Statement.def_around_query_or_exec do |stmt, args, run| - seen << "inner before" + seen.lock { |lines| lines << "inner before" }.or_panic run.call - seen << "inner after" + seen.lock { |lines| lines << "inner after" }.or_panic end defer Statement.clear_around_query_or_exec @@ -382,8 +388,9 @@ describe "Statement" do assert db.scalar("7").or_panic == 7 end - assert seen == ["outer before 1, ? 1", "inner before", "inner after", "outer after", - "outer before 7 0", "inner before", "inner after", "outer after"] + assert seen.get { |lines| lines.join("|") }.or_panic == + ["outer before 1, ? 1", "inner before", "inner after", "outer after", + "outer before 7 0", "inner before", "inner after", "outer after"].join("|") end it "answers an error when a hook does not run the query" do @@ -397,9 +404,9 @@ describe "Statement" do end it "runs nothing once the hooks are cleared" do - count = 0 + count = Atomic(Int32).new(0) Statement.def_around_query_or_exec do |stmt, args, run| - count += 1 + count.add(1) run.call end Statement.clear_around_query_or_exec @@ -407,7 +414,7 @@ describe "Statement" do with_dummy do |db| db.exec("1").or_panic end - assert count == 0 + assert count.get == 0 end end end diff --git a/db/testing/dummy_driver.iyi b/db/testing/dummy_driver.iyi index b64b012..59ced43 100644 --- a/db/testing/dummy_driver.iyi +++ b/db/testing/dummy_driver.iyi @@ -8,16 +8,34 @@ # prepared, `"raise ConnectionLost"` loses the connection, an exec of # `"raise"` or of a connection's `failing_command` fails, and the result # set of `"fail_close"` answers an error when it is closed. +# +# Its connections, statements and result sets may be made and used on any +# worker (`spawn_parallel`): what it counts is atomic, and what it keeps for +# the tests to look at is behind a lock. module db/testing/dummy_driver import std/uri::{URI, Params} import std/text import std/slice::{Slice, Bytes} +import std/mutex::{Mutex} +import std/list::{List} import db import db/errors::* import db/value::* import db/core::* +# A connection or a result set as the double keeps it for the tests to +# look at. `Share` on trust, for this test double alone: a test looks at +# one in the task that used it, or after the tasks that used it have +# ended. +@[Share(owns: true)] +pub class Made(T) + getter value : T + + def initialize(@value : T) + end +end + pub class DummyDriver < Driver def connection_builder(uri : URI) : ConnectionBuilder | DBError params = Params.parse(uri.query || "") @@ -35,8 +53,8 @@ pub class DummyConnectionBuilder < ConnectionBuilder end pub class DummyConnection < Connection - @@connections = [] of DummyConnection - @@connections_count = 0 + @@made : Mutex(Array(Made(DummyConnection))) = Mutex.new { [] of Made(DummyConnection) } + @@connections_count : Atomic(Int32) = Atomic(Int32).new(0) @connected = true @@ -47,20 +65,27 @@ pub class DummyConnection < Connection def initialize(options : Connection::Options) super(options) sleep 0 - @@connections_count += 1 - @@connections << self + @@connections_count.add(1) + made = Made.new(self) + while @@made.lock { |connections| connections << made }.is_a?(Cancelled) + end end def self.connections_count : Int32 - @@connections_count + @@connections_count.get end + # The connections made since `clear_connections`, oldest first. def self.connections : Array(DummyConnection) - @@connections + while true + made = @@made.get { |connections| List(Made(DummyConnection)).new(connections) } + return made.map(&.value) unless made.is_a?(Cancelled) + end end def self.clear_connections : Nil - @@connections = [] of DummyConnection + while @@made.update { |_| [] of Made(DummyConnection) }.is_a?(Cancelled) + end end def build_prepared_statement(query : String) : Statement | DBError @@ -159,28 +184,28 @@ pub class DummySavePointTransaction < SavePointTransaction end pub class DummyStatement < Statement - @@statements_count = 0 - @@statements_exec_count = 0 + @@statements_count : Atomic(Int32) = Atomic(Int32).new(0) + @@statements_exec_count : Atomic(Int32) = Atomic(Int32).new(0) getter params : Hash(Int32, DBValue) def initialize(connection : Connection, command : String, @prepared : Bool) @params = {} of Int32 => DBValue super(connection, command) - @@statements_count += 1 + @@statements_count.add(1) end def self.statements_count : Int32 - @@statements_count + @@statements_count.get end def self.statements_exec_count : Int32 - @@statements_exec_count + @@statements_exec_count.get end protected def perform_query(args : Array(DBValue)) : ResultSet | DBError assert_not_closed - @@statements_exec_count += 1 + @@statements_exec_count.add(1) sleep 0 @connection.as(DummyConnection).check! set_params args @@ -189,7 +214,7 @@ pub class DummyStatement < Statement protected def perform_exec(args : Array(DBValue)) : ExecResult | DBError assert_not_closed - @@statements_exec_count += 1 + @@statements_exec_count.add(1) @connection.as(DummyConnection).check! set_params args if command == "raise" || command == @connection.as(DummyConnection).failing_command @@ -215,7 +240,7 @@ pub class DummyStatement < Statement end pub class DummyResultSet < ResultSet - @@last_result_set : DummyResultSet? = nil + @@last_result_set : Mutex(Made(DummyResultSet)?) = Mutex.new { nil.as(Made(DummyResultSet)?) } @top_values : Array(Array(String)) @values : Array(String)? @@ -227,11 +252,18 @@ pub class DummyResultSet < ResultSet @top_values = command.split.map { |row| row.split(',') } @column_count = @top_values.size > 0 ? @top_values[0].size : 2 @values = nil - @@last_result_set = self + made = Made.new(self).as(Made(DummyResultSet)?) + while @@last_result_set.update { |_| made }.is_a?(Cancelled) + end end def self.last_result_set : DummyResultSet - @@last_result_set || raise "no result set yet" + while true + made = @@last_result_set.get { |last| last } + next if made.is_a?(Cancelled) + return made.value if made + raise "no result set yet" + end end protected def do_close : Nil | DBError diff --git a/manual/pool_concurrency.iyi b/manual/pool_concurrency.iyi index ab8ddd5..8df92bd 100644 --- a/manual/pool_concurrency.iyi +++ b/manual/pool_concurrency.iyi @@ -2,6 +2,9 @@ # # % iyi run manual/pool_concurrency.iyi -- --options="max_pool_size=5" --duration=30 --concurrency=4 # +# The tasks are parallel (`spawn_parallel`), as crystal's fibers were under +# `-Dpreview_mt`: they share one `Database` from every worker, as many +# workers as `IYI_WORKERS` says, or one per core. module manual/pool_concurrency import std/option_parser::{OptionParser} @@ -38,8 +41,7 @@ parser.invalid_option do |flag| end parser.parse -# iyi tasks share one thread (no preview_mt counterpart). -multi_threaded = false +multi_threaded = IyiWorkers.count release = {% if flag?(:release) %} true {% else %} false {% end %} if !release @@ -53,21 +55,22 @@ start_time = Time.monotonic puts "Starting test for #{duration} seconds..." # crystal's free fibers died with the program; iyi's group waits for its -# tasks, so they loop until the duration is over. -running = true +# tasks, so they loop until the duration is over. The flag is an atomic, +# since a parallel task may capture only a `Share` value. +running = Atomic(Int32).new(1) group do |g| concurrency.times do - g.spawn do - while running + g.spawn_parallel do + while running.get == 1 db.scalar "1" sleep 0 end - 0 + nil end end sleep duration * 1000 - running = false + running.set(0) end end_time = Time.monotonic @@ -75,7 +78,7 @@ end_time = Time.monotonic puts " Options : #{options}" puts " Duration (sec) : #{duration} (actual #{end_time - start_time})" puts " Concurrency : #{concurrency}" -puts " Multi Threaded : #{multi_threaded ? "Yes (#{multi_threaded})" : "No"}" +puts " Multi Threaded : Yes (#{multi_threaded} workers)" puts "Total Connections : #{DummyConnection.connections_count}" puts " Total Statements : #{DummyStatement.statements_count}" puts " Total Queries : #{DummyStatement.statements_exec_count}"