rwlocktest.lisp

(eval-when (:compile-toplevel :load-toplevel :execute)
  (ql:quickload :alexandria)
  (ql:quickload :lparallel)
  (ql:quickload :bordeaux-threads)
  (use-package :alexandria))


;;;;;;;;;;;;;;;;;;;;;;;;
;; RWLOCK implementation

;; https://gist.github.com/deadtrickster/5d8202a376309d1e822e

;; Courtesy of https://github.com/deadtrickster

;; https://en.wikipedia.org/wiki/Readers%E2%80%93writers_problem

(defstruct rwlock
  (rlock (bt:make-lock))
  (wlock (bt:make-lock))
  (rtrylock (bt:make-lock))
  (resource (bt:make-lock))
  (rcount 0)
  (wcount 0))

(defun make-rwlock* (&optional (name "RW-LOCK"))
  (make-rwlock :rlock (bt:make-lock (format nil "RLOCK:~A" name))
               :wlock (bt:make-lock (format nil "WLOCK:~A" name))
               :rtrylock (bt:make-lock (format nil "RTRYLOCK:~A" name))
               :resource (bt:make-lock (format nil "RESOURCE:~A" name))))

(defun read-lock-begin (rwlock)
  (bt:acquire-lock (rwlock-rtrylock rwlock))
  (bt:acquire-lock (rwlock-rlock rwlock))
  (when (= 1 (incf (rwlock-rcount rwlock)))
    (bt:acquire-lock (rwlock-resource rwlock)))
  (bt:release-lock (rwlock-rlock rwlock))
  (bt:release-lock (rwlock-rtrylock rwlock)))

(defun read-lock-end (rwlock)
  (bt:acquire-lock (rwlock-rlock rwlock))
  (when (= 0 (decf (rwlock-rcount rwlock)))
    (bt:release-lock (rwlock-resource rwlock)))
  (bt:release-lock (rwlock-rlock rwlock)))

(defmacro with-read-lock (rwlock &body body)
  (with-gensyms (rwlock%)
    `(let ((,rwlock% ,rwlock))
       (read-lock-begin ,rwlock%)
       (unwind-protect
            (progn ,@body)
         (read-lock-end ,rwlock%)))))

(defun write-lock-begin (rwlock)
  (bt:acquire-lock (rwlock-wlock rwlock))
  (when (= 1 (incf (rwlock-wcount rwlock)))
    (bt:acquire-lock (rwlock-rtrylock rwlock)))
  (bt:release-lock (rwlock-wlock rwlock))
  (bt:acquire-lock (rwlock-resource rwlock)))

(defun write-lock-end (rwlock)
  (bt:release-lock (rwlock-resource rwlock))
  (bt:acquire-lock (rwlock-wlock rwlock))
  (when (= 0 (decf (rwlock-wcount rwlock)))
    (bt:release-lock (rwlock-rtrylock rwlock)))
  (bt:release-lock (rwlock-wlock rwlock)))

(defmacro with-write-lock (rwlock &body body)
  (with-gensyms (rwlock%)
    `(let ((,rwlock% ,rwlock))
       (write-lock-begin ,rwlock%)
       (unwind-protect
            (progn ,@body)
         (write-lock-end ,rwlock%)))))

;;;;;;;;;;;;
;; Test case

(defparameter *test-lock* (make-rwlock* "TEST"))

(defvar *test-timeout*) ;; will be set together with other timers in #'start-central-threads

;; Data calculator kernel

(defvar *test-kernel* nil)

(defmacro with-test-kernel ((&key (priority :default) (category :default)) &body body)
  "Call `body' with `*test-kernel' as lparallel kernel"
  `(let ((lparallel:*kernel* *test-kernel*)
         (lparallel:*task-priority* ,priority)
         (lparallel:*task-category* ,category))
     ,@body))

;; Data caclculator channel

(defvar *test-channel* nil)

(defun make-test-channel ()
  (with-test-kernel ()
    ;; Create channel
    (setf *test-channel* (lparallel:make-channel))))

(defun test-start (&optional (workers 2))
  ;; Create kernel
  (unless *test-kernel*
    (setf *test-kernel*
          (lparallel:make-kernel workers :name "LPARALLEL-TEST"))
    ;; Make channel
    (make-test-channel)
    ))

(defun test-compute ()
  "Compute `INDEX-DEFINITION' using data calculator"
  ;; clear the channel result queue
  (loop for retval = (lparallel:try-receive-result *test-channel*)
        while retval
        do (when (typep retval 'condition)
             (error retval)))
  (let ((task (lambda ()
                (with-read-lock *test-lock*
                  (format t "~A: sleeping~%" (bt:current-thread))
                  (sleep 3)
                  (format t "~A: done~%" (bt:current-thread))
                  ))))
    (lparallel:submit-task *test-channel* task)))

(defun test ()
  (test-start)
  (dotimes (x 10)
    (test-compute)))

(defun test2 ()
  (test-start)
  (dotimes (x 10)
    (bt:make-thread
     (lambda ()
       (with-read-lock *test-lock*
         (format t "~A: sleeping~%" (bt:current-thread))
         (sleep 3)
         (format t "~A: done~%" (bt:current-thread))
         )))))

;; Execute test
(test)

This is SBCL 1.4.4.debian, an implementation of ANSI Common Lisp.
More information about SBCL is available at <http://www.sbcl.org/>.

SBCL is free software, provided as is, with absolutely no warranty.
It is mostly in the public domain; some portions are provided under
BSD-style licenses.  See the CREDITS and COPYING files in the
distribution for more information.                          
 
To load "alexandria":
  Load 1 ASDF system:
    alexandria
; Loading "alexandria"

To load "lparallel":
  Load 1 ASDF system:
    lparallel
; Loading "lparallel"

To load "bordeaux-threads":
  Load 1 ASDF system:
    bordeaux-threads
; Loading "bordeaux-threads"

* #<THREAD "LPARALLEL-TEST" RUNNING {1002E94DF3}>: sleeping
#<THREAD "LPARALLEL-TEST" RUNNING {1002E95113}>: sleeping
#<THRTHREAD "LPAARRAALLLEELL--TTEESSTT" RUNRUNGING {1000022EE99541D1F33}}>: done
#<THRTHREAD "LPAARRAALLLEELL--TTEESSTT" RUNRUNGING {1000022EE99541D1F33}}>: done
#<THREAD "LPPARAALLLLELL-TEEST""  RUNNRUNNING {100022EE99541D1F33}}>>: sleeping
#<THREAD "LPPARAALLLLELL-TEEST""  RUNNRUNNING {100022EE99541D1F33}}>>: sleeping
#<THREADTHREAD "LPAARRAALLLLEELL--TTEESSTT"" RUNNING {1002E951D1F33}}>>: done
#<THREADTHREAD "LPAARRAALLLLEELL--TTEESSTT"" RUNNING {1002E951D1F33}}>>: done

debugger invoked on a SB-THREAD:THREAD-DEADLOCK in thread
#<THREAD "LPARALLEL-TEST" RUNNING {1002E94DF3}>:
  Deadlock cycle detected:
    #<SB-THREAD:THREAD "LPARALLEL-TEST" RUNNING {1002E94DF3}>
  waited for:
    #<SB-THREAD:MUTEX "RESOURCE:TEST"
        owner: #<SB-THREAD:THREAD #1="LPARALLEL-TEST" waiting on:
                    #<MUTEX "RTRYLOCK:TEST" owner: #>
                  {1002E95113}>>
  owned by:
    #<SB-THREAD:THREAD #1="LPARALLEL-TEST" waiting on:
         #<MUTEX "RTRYLOCK:TEST"
             owner: #<SB-THREAD:THREAD #1# RUNNING {1002E94DF3}>>
       {1002E95113}>
  waited for:
    (#<SB-THREAD:MUTEX "RTRYLOCK:TEST"
         owner: #<SB-THREAD:THREAD "LPARALLEL-TEST" RUNNING {1002E94DF3}>>)
  owned by:
    #<SB-THREAD:THREAD "LPARALLEL-TEST" RUNNING {1002E94DF3}>

Another attempt (same test case) end with:

o load "alexandria":
 Load 1 ASDF system:
   alexandria
 Loading "alexandria"

o load "lparallel":
 Load 1 ASDF system:
   lparallel
 Loading "lparallel"

o load "bordeaux-threads":
 Load 1 ASDF system:
   bordeaux-threads
 Loading "bordeaux-threads"

 #<THREAD "LPARALLEL-TEST" RUNNING {1002E95113}>: sleeping
LPARALLEL-TEST" RUNNING {1002E94DF3}>: sleeping
<THTHREAD "LPPAARRAALLLLEELL--TTEESSTT""  RUNNING {1002E9945D1F133}}>>: done
<THTHREAD "LPPAARRAALLLLEELL--TTEESSTT""  RUNNING {1002E9945D1F133}}>>: done

ebugger invoked on a SIMPLE-ERROR in thread
<THREAD "LPARALLEL-TEST" RUNNING {1002E94DF3}>:
 Recursive lock attempt #<SB-THREAD:MUTEX "RESOURCE:TEST"
                            owner: #<SB-THREAD:THREAD "LPARALLEL-TEST" RUNNING
                                      {1002E94DF3}>>.

(test2) works just fine:

To load "alexandria":
  Load 1 ASDF system:
    alexandria
; Loading "alexandria"

To load "lparallel":
  Load 1 ASDF system:
    lparallel
; Loading "lparallel"

To load "bordeaux-threads":
  Load 1 ASDF system:
    bordeaux-threads
; Loading "bordeaux-threads"

#<THREAD "Anonymous thread" #<THREAD RUNNING"Anonym o{us thread" 1002F64F73}>: sleeping
RUNNING {1002F65173}>: sleeping
#<THREAD "Anonymous thread" RUNNING {1002F65273}>: sleeping
#<THREAD "Anonymous thread" RUNNING {1002F65273}>: sleeping
ead" 1002F64F73}>: sleeping
RUNNING {1002F65073}>: sleeping
#<THREAD "Anonymous thread" RUNNING {1002F65373}>: sleeping
#<THREAD "Anonymous thread" RUNNING {1002F65473}>: sleeping
#<THREAD "Anonymous thread" RUNNING {1002F65573}>: sleeping
#<THREAD "Anonymous thread" RUNNING {1002F65673}>: sleeping
#<THREAD "Anonymous thread" RUNNING {1002F65773}>: sleeping
#<THREAD "Anonymous thread" RUNNING {1002F65873}>: sleeping
* #<#<THREAD TTHREAD "Anonymou" threa"n ymoouuss  tthhrreeaadd" RUNNING {1002F65173}>RUNNING: done
#<#<THREAD TTHREAD "Anonymou" threa"n ymoouuss  tthhrreeaadd" RUNNING {1002F65173}>RUNNING: done
 {1000022FF66542F7733}}>: done
#<#<THREAD TTHREAD "Anonymou" threa"n ymoouuss  tthhrreeaadd" RUNNING {1002F65173}>RUNNING: done
 {1000022FF66542F7733}}>: done
#<THREAD "Anonymous thread" RUNNING {1002F65373}>: done
#<THREAD "Anonymous thread" RUNNING {1002F65573}>: done
#<THREAD "Anonymous thread" RUNNING {1002F65673}>: done
#<THREAD "Anonymous thread" RUNNING {1002F65773}>: done
#<THREAD "Anonymous thread" RUNNING {1002F65873}>: done
#<THREAD "Anonymous thread" RUNNING {1002F65073}>: done
#<THREAD "Anonymous thread" RUNNING {1002F65473}>: done

So it seems like lparallel is doing something weird what causes this rwlock implementation to fail. 10 regular concurrent bt threads does not conflict with each other, where 2 lparallel workers doing exactly the same thing almost immediately cause deadlock or recursive lock attempt.