Golang: Implement a custom write-preferred Mutex

This is a continuation of series for writing custom synchronization tools provide by Go. Previously we implemented a read-preferred Mutex. We discussed the problem that our implementation is write starving, i.e we could have a scenario where write goroutine is waiting forever.

In this article we will implement write-preferred Mutex, which will give priority to write goroutines over read goroutines. We will also attempt to improve our understanding of how condition variables work in golang.

At the end we will write a sample logging application where multiple routines will be reading and writing our shared log data.

An illustration of how our lock will work. Read op 1&2 are goroutines reading the shared data, while Write op 1&2 are goroutines updating the shared data.

Implementation of the custom write-preferred mutex:

Lets start by writing the struct which will help us visualizing major components of the locking process.

type WPMutex struct {
 totalPendingWriters int // Total amount of writers waiting for lock
 totalActiveReaders  int // Total active readers reading the data
 isAnyWriteActive    bool // Is any write routine active?
 cond                sync.Cond
}

Now lets use our struct to write the two Lock functions. In ReadLock() new read routines will only be allowed to proceed, if no active or pending write routines are present. In WriteLock() we will wait for current read or write routines to finish before proceeding.

func (wm *WPMutex) ReadLock() {
 wm.cond.L.Lock()
 for wm.totalPendingWriters != 0 || wm.isAnyWriteActive {
  wm.cond.Wait()
 }
 wm.totalActiveReaders++
 wm.cond.L.Unlock()
}

func (wm *WPMutex) WriteLock() {
 wm.cond.L.Lock()
 wm.totalPendingWriters++
 for wm.totalActiveReaders != 0 || wm.isAnyWriteActive {
  wm.cond.Wait()
 }
 wm.totalPendingWriters--
 wm.isAnyWriteActive = true
 wm.cond.L.Unlock()
}

In WriteLock(), if three routines A,B and C are waiting. Then only one will be allowed to proceed, this is because wait() function does three steps 1- Unlocks the cond variable. 2- When a signal() is called by some other routine, attempts to gain the lock. 3- When lock is acquired, re-checks the condition in for loop. Hence the first goroutine to gain lock is allowed to proceed, while the other two go back to call wait().

Lets see what happens when our routines have finished their respective processes and now release their locks.

func (wm *WPMutex) ReadUnlock() {
 wm.cond.L.Lock()
 wm.totalActiveReaders--
 if wm.totalActiveReaders == 0 {
  wm.cond.Broadcast()
 }
 wm.cond.L.Unlock()
}

func (wm *WPMutex) WriteUnlock() {
 wm.cond.L.Lock()
 wm.isAnyWriteActive = false
 wm.cond.Broadcast()
 wm.cond.L.Unlock()
}

The ReadUnlock() checks if it was the last active reader then it release our lock for potential write routines. The WriteUnlock() marks that no active routines are present and calls Broadcast() for potential read or write routines that might be waiting.

Following is our implementation for making a new write-preferred lock. Only the condition variables is to be initialized, all other members of WPMutex will have their default values.

func New() *WPMutex {
 return &WPMutex{
  cond: *sync.NewCond(&sync.Mutex{}),
 }
}

Following is a sample logging application that will use the write-preferred lock. There are two routines adding logs and two other routines reading and printing any new logs.

const WriterWait time.Duration = 200 * time.Millisecond
const ReadersProcessingTime time.Duration = 100 * time.Microsecond

type CustomLog struct {
 level   string
 message string
}

func GenerateUserLogs(logsList *[]CustomLog, wm *Writersmutex.WPMutex) {
 sampleLogs := [5]CustomLog{
  {level: "Info", message: "A new user logged-in"},
  {level: "Warn", message: "User entered incorrect credentials"},
  {level: "Info", message: "User added a new order"},
  {level: "Warn", message: "User entered order with no product"},
  {level: "Error", message: "User could not register order"},
 }
 for {
  wm.WriteLock()
  newLog := sampleLogs[rand.Intn(5)]
  *logsList = append(*logsList, newLog)
  wm.WriteUnlock()
  time.Sleep(WriterWait)
 }
}

func GenerateRequestLogs(logsList *[]CustomLog, wm *Writersmutex.WPMutex) {
 sampleLogs := [3]CustomLog{
  {level: "Info", message: "A new request : req-123"},
  {level: "Warn", message: "Order form validation failed: req-456"},
  {level: "Error", message: "Failed process new order: req-159"},
 }
 for {
  wm.WriteLock()
  newLog := sampleLogs[rand.Intn(3)]
  *logsList = append(*logsList, newLog)
  wm.WriteUnlock()
  time.Sleep(WriterWait)
 }
}

func ReportLogs(logsList *[]CustomLog, wm *Writersmutex.WPMutex) {
 processedCount := 0
 for {
  wm.ReadLock()
  if processedCount != len(*logsList) {
   copiedLogs := ExtractNewLogs(*logsList, processedCount)
   processedCount = len(*logsList)
   for _, log := range copiedLogs {
    fmt.Printf("%d - %s: %s\n", processedCount, log.level, log.message)
   }
  }
  time.Sleep(ReadersProcessingTime)
  wm.ReadUnlock()
 }
}

func ReportErrorLogs(logsList *[]CustomLog, wm *Writersmutex.WPMutex) {
 processedCount := 0
 for {
  wm.ReadLock()
  if processedCount != len(*logsList) {
   copiedLogs := ExtractNewLogs(*logsList, processedCount)
   processedCount = len(*logsList)
   for _, log := range copiedLogs {
    if log.level == "Error" {
     fmt.Printf("%d - ***An error log*** %s: %s\n", processedCount, log.level, log.message)
    }
   }
  }
  time.Sleep(ReadersProcessingTime)
  wm.ReadUnlock()
 }
}

func ExtractNewLogs(cLog []CustomLog, processedCount int) []CustomLog {
 copiedCLog := make([]CustomLog, 0)
 for i := processedCount; i < len(cLog); i++ {
  copiedCLog = append(copiedCLog, cLog[i])
 }
 return copiedCLog
}

func main() {
 logsList := make([]CustomLog, 0)
 readerMutex := Writersmutex.New()
 go ReportLogs(&logsList, readerMutex)
 go ReportErrorLogs(&logsList, readerMutex)
 go GenerateUserLogs(&logsList, readerMutex)
 GenerateRequestLogs(&logsList, readerMutex)
}

If we had used our read-preferred mutex developed in the previous article then no log will be outputted since now our read routines run continuously and will always gain access to the write lock.

Conclusion:

In this article we implemented a new type of Mutex which gives priority to routines that are writing the data, while not providing exclusive lock access to read routines. We also used the newly created Mutex in a sample application.

The full source code for Mutex and sample application can be found on Github link.