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)
#
# -----------------------------------------------------------------------

# blocking_mutate -- a variant of mutate that may be shared between threads
#
# This mutate ensure exclusive access to shared memory via using locks and
# blocking.
#
# Accidental race accesses are not permitted by this mutate, i.e., accessing
# mutable elements, e.g, via `get` or `put` on a mutable var `mutate.new T`,
# are not permitted unless we are in code run via
# `blocking_mutate.this.exclusive`.
#
public blocking_mutate : mutate is

  # mutex to be used for synchronization
  #
  mtx := concur.sync.mutex.new.or_else
            (panic "`concur.sync.mutex.new` failed to create mutex for `concur.blocking_mutate`")


  # 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.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.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.this.replace

        res := code()

        wakeup_next_in_waiting_list

        set exclusive_thread := nil
        blocking_mutate.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  # NYI: result ignored
      _ := mtx.lock    # NYI: result ignored,
      set exclusive_thread := concur.Threads.env.current


  # is instance `mutate.this` instated for type `mutate.this` and, if `mutate.this` does
  # not support use in several threads, check we are running in the thread `mutate.this`
  # was created in?
  #
  public redef is_this_mutate_instated_and_usable bool
  =>
    blocking_mutate.this.get_if_instated >>? (x -> id = x.id)


  # 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