An Optimistic Concurrency Control Manager

I took a few days off work this week and wanted to write some fun code while I was out. Learning more about concurrency management has been on my list for a while, so let’s go with that. Broadly, databases tend to manage concurrency with either

\textbf{Pessimistic}

(e.g. Two-Phase Locking) or

\textbf{Optimistic}

(e.g. Timestamp Ordering) protocols. In a textbook implementation of OCC, a DB is made serializable by taking no locks during a preliminary working phase and then write-locking the DB during a short validation and write phase.

Today, I roughly implemented Kung and Robinson’s On Optimistic Methods for Concurrency Control (1981) and wrote a small (

\leq 200

LOC) Go package that can be used to wrap an arbitrary data store and give it OCC semantics. Because direct ports of papers are hardly ever fun, I made the following changes:

In the example below I initialize a

\texttt{noopStore}

which satisfies

\texttt{occ.Store}

and use it to manage several dozen concurrent readers and writers. Acknowledging that this is a trivial example, we’re looking at the ability to verify several million operations per second.

package main

import (
    "math/rand"
    "sync"
    "occ"
)

type noopStore int

func (s noopStore) Write(ds []interface{}) error {
    return nil
}

func (s noopStore) Read(id uint64) (interface{}, error) {
    return 1, nil
}

func (s noopStore) Merge(old, new interface{}) (interface{}, error) {
    return 1, nil
}

func main() {

    var (
        numEvents, numWorkers  int    = 1000000, 48
        hashBits, txnRetention uint64 = 1024, 10000
        numObjects             uint64 = 1024 * 1024
        workCh                        = make(chan uint64)
        wg                            = sync.WaitGroup{}
    )

    DB := occ.NewManager(noopStore(1), hashBits, txnRetention)

    wg.Add(numWorkers)
    for workerID := 0; workerID < numWorkers; workerID++ {
        go func() {
            defer wg.Done()
            c := DB.NewConnection()
            for r := range workCh {
                c.Begin()
                c.Read(r)
                c.Write(r, 1)
                c.Commit()
            }
        }()
    }

    for i := 0; i < numEvents; i++ {
        workCh <- (rand.Uint64() % numObjects)
    }
    close(workCh)
    wg.Wait()

    // A little contrived example, `commitRate` subject to fluctuate on quite a few factors, 
    // number of items read,  max concurrency, hashBits, variation in exec time, etc. This 
    // is as favorable as it gets...
    // 
    // 2024/08/23 13:38:20 commitRate: 0.994779, execTime: 387.828209ms

}