NAME
Acme::Parataxis::TVar - A versioned cell whose reads and writes commit all at once
SYNOPSIS
use v5.40;
use Acme::Parataxis qw[async fiber await_sleep atomically retry];
use Acme::Parataxis::TVar;
my $account = Acme::Parataxis::TVar->new( value => 5 );
async {
my $taker = fiber {
atomically {
my $have = $account->get;
retry if $have < 10; # not enough yet: park until there is
$account->set( $have - 10 );
return 'took 10';
}
};
await_sleep( 10 );
say 'parked, and the account reports ', $account->waiters, ' waiter(s)';
atomically { $account->set( $account->get + 20 ) };
say $taker->await, ', leaving ', $account->value;
};
# parked, and the account reports 1 waiter(s)
# took 10, leaving 15DESCRIPTION
A TVar is a mutable cell with a version number. You do not read and write it directly; you open a transaction around the work with atomically and go through get and set. Nothing takes effect until the outermost transaction commits, and when it does, every write in it lands together or none of them does.
That is the whole point. Two fibers cannot interleave halfway through a pair of writes, because the pair is not two writes - it is one commit. A transfer that reads two accounts and writes two accounts is atomic as a unit, so a second fiber working on the same accounts either sees it whole or not at all. There is no lock to take and nothing to release.
The cost is that the block may not run once. It is re-run whenever a TVar it read was committed by someone else in the meantime, and it is abandoned and re-run from the top when it calls retry. A transaction is therefore not a place for side effects you cannot repeat: no printing, no file handles, no put on a channel. "The block may run many times" is the habit that follows from that.
use v5.40;
use Acme::Parataxis qw[async fiber await_sleep atomically];
use Acme::Parataxis::TVar;
my $a = Acme::Parataxis::TVar->new( value => 100 );
my $b = Acme::Parataxis::TVar->new( value => 0 );
async {
my @jobs = map {
fiber {
atomically {
my $from = $a->get;
my $to = $b->get;
await_sleep( 2 );
$a->set( $from - 25 );
$b->set( $to + 25 );
return 1;
}
}
} 1 .. 4;
$_->await for @jobs;
say 'a = ', $a->value, ', b = ', $b->value;
};
# a = 0, b = 100Four fibers each moved 25 out of the same account, and every 25 arrived. No fiber saw a half-finished transfer, and the total is unchanged.
METHODS
new( value = $v )>
use v5.40;
use Acme::Parataxis::TVar;
my $count = Acme::Parataxis::TVar->new( value => 0 );
say 'starting at ', $count->value, ', version ', $count->version;
my $text = Acme::Parataxis::TVar->new( value => 'anything' );
say 'the value is not type-checked: ', $text->value;
# starting at 0, version 0
# the value is not type-checked: anythingCreates a cell holding $v. value is required - there is no sensible default - and it is stored as given, so any scalar will do. A new cell is at version 0.
version is not a constructor parameter, and neither is anything else. The check for an unrecognised key never gets a chance to run when value is missing, because that is reported first.
use v5.40;
use Acme::Parataxis::TVar;
for my $args ( [ 'value' => 1 ], [ 'version' => 3 ], [] ) {
my $label = @$args ? join ' ', @$args : 'nothing at all';
my $built = eval { Acme::Parataxis::TVar->new( @$args ); 'ok' };
say "new( $label ): ", ( $@ ? 'croaks' : $built );
}
# new( value 1 ): ok
# new( version 3 ): croaks
# new( nothing at all ): croaksget( )
use v5.40;
use Acme::Parataxis qw[async atomically];
use Acme::Parataxis::TVar;
my $cell = Acme::Parataxis::TVar->new( value => 10 );
async {
say 'read: ', atomically { $cell->get };
say 'and the version is still ', $cell->version;
say 'read-your-writes: ', atomically { $cell->set( 5 ); $cell->get };
say 'once committed, the version is ', $cell->version;
};
# read: 10
# and the version is still 0
# read-your-writes: 5
# once committed, the version is 1Reads the cell and journals the read against the enclosing transaction. Off a scheduled fiber it croaks with TVar operations must occur inside a scheduled fiber; on one but outside atomically it croaks with TVar operations must occur inside an atomically block.
Two things make this different from value. It reads your own writes, so a set earlier in the same transaction is what a later get returns. And the read is journaled, which is what makes the transaction safe: if anyone commits this cell between your read and your commit, the whole block is re-run rather than allowed to write on top of a value it has not seen.
set( $v )
use v5.40;
use Acme::Parataxis qw[async atomically];
use Acme::Parataxis::TVar;
my $cell = Acme::Parataxis::TVar->new( value => 10 );
async {
my $given = atomically { $cell->set( 'replaced' ) };
say 'set returns the value it was given: ', $given;
say 'and the cell holds ', $cell->value;
for my $v ( 0, '', undef ) {
my $falsy = Acme::Parataxis::TVar->new( value => 1 );
atomically { $falsy->set($v) };
my $shown = defined $falsy->value ? "'" . $falsy->value . "'" : 'undef';
say 'a false value commits fine: ', $shown;
}
};
# set returns the value it was given: replaced
# and the cell holds replaced
# a false value commits fine: '0'
# a false value commits fine: ''
# a false value commits fine: undefStages $v as this transaction's new value and returns it. The committed value is untouched until the outermost transaction commits, so a set inside the block is invisible to value and to every other fiber until then.
set and get have the same two guards, and neither is special about falsy values - 0, the empty string and undef all commit as themselves.
value and version
use v5.40;
use Acme::Parataxis qw[async atomically];
use Acme::Parataxis::TVar;
my $cell = Acme::Parataxis::TVar->new( value => 3 );
say 'before anything: ', $cell->value, ' at version ', $cell->version;
async {
atomically { $cell->set( 4 ) };
say 'after one commit: ', $cell->value, ' at version ', $cell->version;
atomically { $cell->set( 4 ) };
say 'and after setting the same value again: ', $cell->value, ' at version ', $cell->version;
};
# before anything: 3 at version 0
# after one commit: 4 at version 1
# and after setting the same value again: 4 at version 2value is the committed value, readable from anywhere with no transaction and no fiber. It is for inspection - tests, logging, reporting - and it is not a transactional read: it does not journal anything, so a set computed from it can silently overwrite someone else's work. "value is not a transactional read" has the demonstration, and it is the single easiest way to get this module wrong.
version is the conflict detector's input, bumped by every commit whether or not the value actually changed. Both are readers, so both work off a scheduled fiber as well as on one.
waiters
use v5.40;
use Acme::Parataxis qw[async fiber await_sleep atomically retry];
use Acme::Parataxis::TVar;
my $queue = Acme::Parataxis::TVar->new( value => 0 );
async {
my $hungry = fiber {
atomically {
my $have = $queue->get;
retry if $have == 0; # park until the cell is written
return "woke up and found $have";
}
};
await_sleep( 10 );
say 'one fiber is parked in retry, so waiters is ', $queue->waiters;
atomically { $queue->set( 1 ) };
say $hungry->await;
say 'and after the commit that woke it, waiters is ', $queue->waiters;
};
# one fiber is parked in retry, so waiters is 1
# woke up and found 1
# and after the commit that woke it, waiters is 0How many fibers are parked in retry with this cell in their read set. It is a plain method and works off a fiber, and like waiters elsewhere in Acme::Parataxis::Sync it drops the moment the queue is served rather than when the waiters run.
Transactions
atomically and retry are not TVar methods. They are exported from Acme::Parataxis and a TVar is what they operate on, so they are documented here because you cannot use this class without them.
atomically( sub { ... } )
use v5.40;
use Acme::Parataxis qw[async fiber await_sleep atomically];
use Acme::Parataxis::TVar;
my $a = Acme::Parataxis::TVar->new( value => 0 );
my $b = Acme::Parataxis::TVar->new( value => 0 );
async {
atomically {
$a->set( 1 );
my $inner = atomically { $b->set( 2 ); 'the inner block ran' };
say $inner;
say 'but b is not committed yet: ' . $b->value . ' at version ' . $b->version;
};
say 'now it is: b = ' . $b->value . ' at version ' . $b->version;
};
# the inner block ran
# but b is not committed yet: 0 at version 0
# now it is: b = 2 at version 1Runs the block as one transaction on the calling fiber and returns whatever the block returns, in the caller's context. Off a scheduled fiber it croaks with atomically() must be called from inside a scheduled fiber.
A nested atomically does not start a second transaction - it joins the one already running. The inner block's writes go into the same write set, and the two commit together at the outer boundary. There is no way to commit part of a transaction.
The sub has a (&) prototype, so atomically { ... } and atomically( sub { ... } ) are both fine and Acme::Parataxis->atomically( sub { ... } ) works too. What you cannot do is pass a non-CODE: that is caught by the parser rather than at runtime. The prototype also means the block form needs a terminating semicolon if any statement follows it - "A block form needs its semicolon" explains why.
retry( )
use v5.40;
use Acme::Parataxis qw[async fiber await_sleep atomically retry];
use Acme::Parataxis::TVar;
my $queue = Acme::Parataxis::TVar->new( value => 0 );
my $runs = 0;
async {
my $worker = fiber {
atomically {
$runs++;
retry if $queue->get == 0; # nothing to do: come back later
$queue->set( $queue->get - 1 );
return 'a worker took one';
}
};
await_sleep( 10 );
atomically { $queue->set( $queue->get + 1 ) };
say $worker->await, ', after ', $runs, ' run(s) of the block';
};
# a worker took one, after 2 run(s) of the blockAbandons the current transaction, parks the fiber until one of the TVars it read is committed, and then re-runs the block from the top. It is how you wait for a condition without polling and without holding anything.
There are three ways to get it wrong, and all three are checked. Off a scheduled fiber it croaks with retry() must be called from inside a scheduled fiber. Outside a transaction it croaks with TVar operations must occur inside an atomically block. And with nothing read it croaks with retry with an empty read set would never wake; read at least one TVar before retry() - which is the useful one, because the cell it names as the reason is exactly the cell the caller has to read to have a condition to wait on.
retry inside a nested atomically aborts the whole transaction, not just the inner block. See "A nested retry discards the outer block".
How It Works
The state is two values and a list. value is what everyone else sees, version counts the commits, and waiters is the fiber ids currently parked on this cell.
The transaction itself is not on the TVar at all. It lives in a slot on the calling fiber's locals stash, keyed so it cannot collide with a Acme::Parataxis::Local id, and it holds two maps and a list. The maps are keyed by the cell's reference address, so a transaction can span any number of cells; the list is the cells in the order they were first read, and it is what a retry parks on.
get and set are the only two methods that touch the log, and they are deliberately asymmetric:
setwrites into the write set and returns. It never journals a read, so a transaction that only writes cannot conflict. "value is not a transactional read" is the same trap reached from the other side.getlooks in the write set first, so a transaction reads its own writes. Failing that it records[ $cell, $version ]- but only the first time, because a secondgetof the same cell has nothing new to protect.
Commit is the interesting part, and it is three steps under a single global mutex. Take the lock, check that every journaled read still carries the version it had when it was read, and only then flush the write set. Each flush publishes the value, bumps that cell's version, and wakes the fibers parked on it. The mutex is never contended in a cooperative scheduler because the commit never yields.
All-or-nothing is what makes this deadlock-proof. Two transactions that want the same two cells cannot be half-applied and cannot wait for each other: one takes the commit lock, the other either finds a stale read and re-runs or waits for the lock and then finds a stale read. Either way somebody re-runs, and a re-run recomputes from committed values rather than resuming from a broken intermediate state.
A failed check is not an error. _atomically wipes the log and runs the block again, which is why the block has to be safe to repeat. retry takes the other exit: it throws an internal control-flow exception, the transaction parks on its read set, and the loop starts over with a fresh log.
Recipes
Moving a value between two cells
The shape the module exists for. Read both cells, do the work, write both back - and the pair is committed together, so no fiber ever sees one account debited without the other credited.
use v5.40;
use Acme::Parataxis qw[async fiber await_sleep atomically];
use Acme::Parataxis::TVar;
my $wallet = Acme::Parataxis::TVar->new( value => 100 );
my $savings = Acme::Parataxis::TVar->new( value => 0 );
async {
my @depositors = map {
fiber {
atomically {
my $from = $wallet->get;
my $to = $savings->get;
await_sleep( 2 );
$wallet->set( $from - 20 );
$savings->set( $to + 20 );
return 1;
}
}
} 1 .. 3;
$_->await for @depositors;
say 'wallet = ', $wallet->value . ', savings = ', $savings->value
. ', total = ' . ( $wallet->value + $savings->value );
};
# wallet = 40, savings = 60, total = 100The await_sleep is there on purpose: it opens the window in which the three transactions collide, so the run really does exercise the re-run path rather than getting lucky.
A read-modify-write that cannot lose an update
Incrementing a counter is the smallest thing that can go wrong, and the difference between doing it with get and doing it with value is the whole module.
use v5.40;
use Acme::Parataxis qw[async fiber await_sleep atomically];
use Acme::Parataxis::TVar;
my $counter = Acme::Parataxis::TVar->new( value => 0 );
my $runs = 0;
async {
my @workers = map {
fiber {
atomically {
$runs++;
my $seen = $counter->get; # the transactional read
await_sleep( 5 );
$counter->set( $seen + 1 );
return 1;
}
}
} 1 .. 4;
$_->await for @workers;
say 'the counter is ', $counter->value, ', and the block ran ', $runs, ' time(s)';
};
# the counter is 4, and the block ran 10 time(s)Ten runs for four increments is normal, not a defect: every one of those fibers reads and writes the same cell, so each commit is checked against the one before it and the losers start again. The counter is right, which is the only thing that matters - but a cell that every fiber fights over is a contention point, and the retry count is the number you watch to find it.
Taking work only when there is some
A work queue as a single cell holding a count. A worker that finds it empty does not spin and does not block forever on a lock; it parks on the cell itself and is woken by the commit that makes work available.
use v5.40;
use Acme::Parataxis qw[async fiber await_sleep atomically retry];
use Acme::Parataxis::TVar;
my $queue = Acme::Parataxis::TVar->new( value => 1 ); # one job waiting up front
my @log;
async {
my @workers = map {
my $id = $_;
fiber {
my $attempts = 0;
my $got = atomically {
$attempts++;
retry if $queue->get == 0; # nothing to do: park on the cell
$queue->set( $queue->get - 1 );
return "worker $id took one, on attempt $attempts";
};
push @log, $got;
}
} 1 .. 2;
await_sleep( 10 );
push @log, 'one worker is parked, so the queue reports ' . $queue->waiters . ' waiter(s)';
atomically { $queue->set( $queue->get + 1 ) };
$_->await for @workers;
push @log, 'the queue is left holding ' . $queue->value;
};
say for @log;
# worker 1 took one, on attempt 1
# one worker is parked, so the queue reports 1 waiter(s)
# worker 2 took one, on attempt 2
# the queue is left holding 0Note where the retry is: after the get, inside the same transaction. That is what puts the cell in the read set, and therefore what gives the park something to be woken by. A retry before any read is the empty-read-set croak from "retry( )".
Behavior Notes
A block form needs its semicolon
atomically { ... } hands the braces to the sub as its block argument, and Perl then keeps collecting arguments after the closing brace until it reaches something that cannot be part of a call. A statement on the next line is exactly such a thing, so the block runs on as an argument list and the compile fails with a message that names the wrong culprit: Too many arguments for Acme::Parataxis::atomically.
use v5.40;
use Acme::Parataxis qw[async atomically];
use Acme::Parataxis::TVar;
my $cell = Acme::Parataxis::TVar->new( value => 1 );
async {
atomically { $cell->set( 2 ) };
say 'committed, and the value is ', $cell->value;
};
# committed, and the value is 2The semicolon after the closing brace is what ends the call. Delete it and the next line is where the error is reported, whatever that line happens to be. This is a rule of the (&) prototype rather than of this module - any sub taking a block behaves the same way - and it applies just as much to async and fiber. atomically( sub { ... } ) is immune, and so is any block that is the last thing in its enclosing block.
value is not a transactional read
value is readable from anywhere, which makes it the natural thing to reach for. It does not journal the read, so nothing detects a conflict, and a set computed from it overwrites whatever landed in between.
use v5.40;
use Acme::Parataxis qw[async fiber await_sleep atomically];
use Acme::Parataxis::TVar;
my $counter = Acme::Parataxis::TVar->new( value => 0 );
async {
my @workers = map {
fiber {
atomically {
my $seen = $counter->value; # readable, but not journaled
await_sleep( 5 );
$counter->set( $seen + 1 );
return 1;
}
}
} 1 .. 4;
$_->await for @workers;
say 'four increments, and the counter reads ', $counter->value;
};
# four increments, and the counter reads 1Every worker read 0, every worker wrote 1, and three of the four writes were thrown away - silently, with no diagnostic, because from the transaction's point of view nothing conflicted. There is no version of this that is merely unlucky: a write-only transaction has an empty read set, and an empty read set always validates.
The rule is short. Inside a transaction, read with get. Outside one, read with value and do not write.
The block may run many times
A transaction is a candidate for re-execution, not a promise of one run. It re-runs on a commit conflict, and retry abandons it and starts again from the first line. Anything the block does outside the cells is repeated with it: printing, logging, appending to an array, sending on a channel, spending real money.
use v5.40;
use Acme::Parataxis qw[async fiber await_sleep atomically];
use Acme::Parataxis::TVar;
my $counter = Acme::Parataxis::TVar->new( value => 0 );
my @side_effects;
async {
my @workers = map {
fiber {
atomically {
my $seen = $counter->get;
await_sleep( 5 );
$counter->set( $seen + 1 );
push @side_effects, 'charged'; # once per run, not once per worker
return 1;
}
}
} 1 .. 4;
$_->await for @workers;
say 'the counter is ', $counter->value, ' but ', scalar @side_effects, ' charges were made';
};
# the counter is 4 but 10 charges were madeThe cell is correct and the ledger is not, and the ledger is the one a customer sees. Compute the new value inside the transaction and perform the effect after it, from the committed state.
A nested retry discards the outer block
atomically blocks nest into one transaction, and retry aborts the transaction. A retry therefore reaches further than the block it is written in: any write the enclosing code had staged is thrown away with it.
use v5.40;
use Acme::Parataxis qw[async fiber await_sleep atomically retry];
use Acme::Parataxis::TVar;
my $gate = Acme::Parataxis::TVar->new( value => 0 );
my $side = Acme::Parataxis::TVar->new( value => 0 );
my $runs = 0;
async {
my $opener = fiber {
await_sleep( 10 );
atomically { $gate->set( 1 ) };
};
atomically {
$runs++;
unless ( $gate->get ) {
$side->set( $side->get + 1 ); # staged, then abandoned
atomically { retry() };
}
$side->set( $side->get + 1 ); # this one commits
};
$opener->await;
say 'the block ran ', $runs, ' time(s) and the side effect landed as ', $side->value;
};
# the block ran 2 time(s) and the side effect landed as 1That is usually what you want, and it is worth knowing deliberately rather than discovering. The re-run starts at the top of the outermost atomically, so the code before the nested retry runs again from the beginning.
Reading one cell twice is not a snapshot
Only the first get of a cell in a transaction journals its version, so a second get returns whatever the cell holds at that moment - which may be a commit the transaction has not accounted for. The commit check still catches it, because the version it recorded has moved, so the block is re-run and the newer value is discarded rather than used.
use v5.40;
use Acme::Parataxis qw[async fiber await_sleep atomically];
use Acme::Parataxis::TVar;
my $cell = Acme::Parataxis::TVar->new( value => 0 );
my $runs = 0;
async {
my $other = fiber {
await_sleep( 10 );
atomically { $cell->set( 99 ) };
};
my $seen = atomically {
$runs++;
my $first = $cell->get;
$other->await;
my $second = $cell->get;
return "$first then $second";
};
say "the block saw $seen, ran $runs time(s), and committed ", $cell->value;
};
# the block saw 99 then 99, ran 2 time(s), and committed 99So this is not a correctness trap - the re-run is what makes it safe. It does mean a second get is not a stable read, and that a transaction which reads a cell it has already read is doing more work than it looks.
An error discards the write, and leaves the cell usable
An exception inside a transaction propagates unchanged and takes the staged writes with it. Nothing is committed, no retry is attempted, and the cell is exactly as it was.
use v5.40;
use Acme::Parataxis qw[async atomically];
use Acme::Parataxis::TVar;
my $cell = Acme::Parataxis::TVar->new( value => 1 );
async {
eval {
atomically {
$cell->set( 2 );
die "the rule was rejected\n";
}
};
my $why = $@;
$why =~ s/ at .*//s;
say "the error came out intact: $why";
say 'nothing was committed: value ', $cell->value, ', version ', $cell->version;
say 'and the next transaction is clean: ', atomically { $cell->get };
};
# the error came out intact: the rule was rejected
# nothing was committed: value 1, version 0
# and the next transaction is clean: 1The same holds for a deadline. A with_timeout that fires inside a transaction aborts it, and the log is cleared on the way out, so the next atomically on that fiber starts fresh rather than inheriting a half-built transaction.
A retry park is a real park
retry parks the fiber on an ordinary fiber wait, which means the scheduler's deadlock detection applies to it. A fiber waiting for work that will never be queued is indistinguishable from a deadlock, and is reported as one.
That is the correct outcome, and it is a loud one: the alternative - a retry loop that silently spins - is what you would be trying to avoid. But it does mean a worker pool needs a way to stop. Either bound the wait with with_timeout, or publish a sentinel on the cell that every waiting worker can get and act on.
SEE ALSO
Acme::Parataxis, Acme::Parataxis::Sync, Acme::Parataxis::Sync::Mutex, Acme::Parataxis::Sync::RwLock, Acme::Parataxis::Error::Timeout, Acme::Parataxis::Error::Cancelled
AUTHOR
Sanko Robinson https://github.com/sanko
LICENSE
Copyright (C) Sanko Robinson.
This library is free software; you can redistribute it and/or modify it under the terms found in the Artistic License 2.