The pool and Database are safe and Share across workers

.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
 

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

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

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