concur/blocking_mutate.fz

# This file is part of the Fuzion language implementation.
#
# The Fuzion language implementation is free software: you can redistribute it
# and/or modify it under the terms of the GNU General Public License as published
# by the Free Software Foundation, version 3 of the License.
#
# The Fuzion language implementation is distributed in the hope that it will be
# useful, but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU General Public
# License for more details.
#
# You should have received a copy of the GNU General Public License along with The
# Fuzion language implementation.  If not, see <https://www.gnu.org/licenses/>.


# -----------------------------------------------------------------------
#
#  Tokiwa Software GmbH, Germany
#
#  Source code of Fuzion standard library feature blocking_mutate
#
#  Author: Fridtjof Siebert (siebert@tokiwa.software)
#
# -----------------------------------------------------------------------

# Workaround for #7638, intermediate feature that is a parent only of
# blocking_mutate_using_clock.
#
module blocking_mutate_using_clock0 : mutate is


# blocking_mutate_using_clock -- a multi-thread variant of mutate using a given Clock
#
# This mutate ensures exclusive access to shared memory via using locks and
# blocking.
#
# Accidental racy accesses are not permitted by this mutate, i.e., accessing
# mutable elements, e.g, via `get` or `put` on a mutable var
# `blocking_mutate_using_clock.this.env.new T`, are not permitted unless we are
# in code run via `blocking_mutate.this.env.exclusive`.
#
# For timeouts, this uses the given clock effect.
#
public blocking_mutate_using_clock(C type : time.Clock) : blocking_mutate_using_clock0 is

  # mutex to be used for synchronization
  #
  mtx := concur.sync.mutex.new.or_panic


  # Threads that are currently blocked in `wait`
  #
  waiting := concur.Thread_List


  # thread that currently has exclusive access to this mutate.
  #
  # This is usually `nil`, unless we are in an exclusive section.
  #
  # Note that from the point of view of a thread `T1` the value of
  # `blocking_mutate_using_clock.this.env.exclusive_thread` may be `nil` while
  # this value for another thread `T2` may be equal to `T2` since this is a
  # value type feature and setting this in one thread usually has no effect on
  # the value seen in other threads.
  #
  # There is one exception, though: If a new thread `T3` is spawned off
  # from `T1` in an exclusive section, then `T3` will receive a copy of
  # `T1`'s `blocking_mutate_using_clock.this.env` where `exclusive_thread` is `T1`.
  # Apart from this case, `exclusive_thread` should always be either `nil` or, if
  # in an exclusive section, the `concur.Threads.env.current`.
  #
  module redef exclusive_thread option concur.Thread := nil


  # perform given code with exclusive access to the mutable values created with
  # this instance of `mutate`.
  #
  public redef exclusive(R type, F type: ()->R, code F) R
  =>
    me := concur.Threads.env.current
    if exclusive_thread = me then
      # recursive `exclusive` section, so we can just run `code`:
      code()
    else
      # we must synchronize first:
      mtx.synchronized ()->
        set exclusive_thread := me
        blocking_mutate_using_clock.this.replace

        res := code()

        wakeup_next_in_waiting_list

        set exclusive_thread := nil
        blocking_mutate_using_clock.this.replace

        res


  # before we leave an exclusive section, check if the wait condition
  # of any waiting thread evaluates to true and, if so, wake up that
  # thread.
  #
  wakeup_next_in_waiting_list
  pre
    debug: is_exclusive
  =>
    match waiting.first
      f concur.Thread =>
        for t := f, nxt
            nxt := t.next
        while !t.park_condition()
        until nxt = f  # we went once around the linked list but found nothing
        else
          # t.park_condition() is true, so unpark `t`.
          waiting.remove t
          t.unpark
      nil =>


  # wait for the given condition to become true by a mutation of the underlying data.
  #
  # In case this mutate is not a multi-thread implementation and `condition()` is `false`,
  # this will `panic`.
  #
  public redef wait(F type: ()->bool, condition F) unit
  =>
    while !condition() do
      wakeup_next_in_waiting_list
      concur.Threads.env.current.park condition ()->
        waiting.add concur.Threads.env.current
        mtx.unlock
      mtx.lock
      set exclusive_thread := concur.Threads.env.current


  # Perform a wait or exclusive_when with a given timeout
  #
  # This allows perform `wait` or `exclusive_when` with a timeout as follows:
  #
  #     mutate_type.env.exclusive ()->
  #       if mutate_type.env.timeout t
  #                         .wait condition
  #       then
  #         # condition was met before timeout
  #       else
  #         # timeout occurred before condition became true
  #
  # or
  #
  #       match mutate_type.env.timeout t
  #                            .exclusive_when condition ()->
  #                              # condition was met before timeout
  #         nil => # timeout occurred before condition became true
  #         r T => # r was returned by code passed ot `exclusive_when`.
  #
  public timeout(deadline time.duration | time.instant)
  is

    # convert relative duration to absolute instant:
    timeout := match deadline
      d time.duration => C.env.read + d
      i time.instant  => i

    # wait for the given condition to become true by a mutation of the underlying data.
    #
    public wait(F type: ()->bool, condition F) bool
    pre
      safety: is_exclusive
    post
      debug: (result: condition())
    =>
      while !condition() do
        wakeup_next_in_waiting_list
        wokeup := concur.Threads.env.current.park_until C _ timeout condition ()->
          waiting.add concur.Threads.env.current
          mtx.unlock
        mtx.lock
        set exclusive_thread := concur.Threads.env.current
      until !wokeup  # -- timeout
        false
      else
        true

    # exclusively wait for the given condition to hold and then perform given code with
    # exclusive access to the mutable values created with this instance of `mutate`.
    #
    # In case of a timeout, the result will be nil. Otherwise, the result of running `code()`
    # will be wrapped in an option and returned.
    #
    public exclusive_when(R type,
                          O type: ()->bool,
                          F type: ()->R,
                          condition O,
                          code F
                          ) option R
    pre
      safety: exclusive_allowed
    =>
      blocking_mutate_using_clock.this.env.exclusive ()->
        if timeout.this.wait condition
          option code()
        else
          nil


  # cleanup feature to destroy mutex and other resources associated with this
  # effect.
  #
  # NYI: UNDER DEVELOPMENT: Ensure that this does not get redefined by user
  # code.
  #
  public redef finally unit
  =>
    mtx.destroy


# blocking_mutate -- a multi-thread variant of mutate using time.Monotonic
#
# This mutate ensures exclusive access to shared memory via using locks and
# blocking.
#
# Accidental racy accesses are not permitted by this mutate, i.e., accessing
# mutable elements, e.g. via `get` or `put` on a mutable var
# `blocking_mutate.this.env.new T`, are not permitted unless we are in code run via
# `blocking_mutate.this.env.exclusive`.
#
public blocking_mutate : blocking_mutate_using_clock time.Monotonic is