Search⌘ K
Home
Courses
The Art of PostgreSQL

Putting Concurrency to the Test

Explore how PostgreSQL manages concurrency through isolation levels and locking mechanisms in transaction processing. Understand the impact of unique constraints on concurrency, analyze benchmark tests comparing insert and update operations under high concurrent loads, and learn strategies to improve scalability when multiple users access shared resources simultaneously.

We'll cover the following...

When we benchmark the concurrency properties of the two statements, we quickly realize that the activity table is badly designed. The unique constraint includes a timestamptz field, which in PostgreSQL is only precise down to the microsecond.

Fixing the table

This kind of made-up unique constraint means we now have these errors to deal with:

Error: Database error 23505: duplicate key value violates unique constraint "activity_messageid_datetime_action_key"
DETAIL: Key (messageid, datetime, action)=(1, 2022-06-01 10:17:39.079724+00, rt) already exists.

The best course of action here is to do this:

PostgreSQL
alter table tweet.activity
drop constraint activity_messageid_datetime_action_key;

Comparing the concurrency scaling

Now, we can properly compare the concurrency scaling of the insert and the update-based versions. In case you might be curious about it, here’s the testing code that’s been used:

XML
(defpackage #:concurrency
  (:use #:cl #:appdev)
  (:import-from #:lparallel
                #:*kernel*
                #:make-kernel #:make-channel
                #:submit-task #:receive-result
                #:kernel-worker-index)
  (:import-from #:cl-postgres-error
                #:database-error)
  (:export      #:*connspec*
                #:concurrency-test))
(in-package #:concurrency)
(defparameter *connspec* '("appdev" "dim" nil "localhost"))
(defparameter *insert-rt*
  "insert into tweet.activity(messageid, action) values($1, 'rt')")
(defparameter *update-rt*
  "update tweet.message set rts = coalesce(rts, 0) + 1 where messageid = $1")
(defun concurrency-test (workers retweets messageid
                         &optional (connspec *connspec*))
  (format t "Starting benchmark for updates~%")
  (with-timing (rts seconds)
      (run-workers workers retweets messageid *update-rt* connspec)
    (format t "Updating took ~f seconds, did ~d rts~%" seconds rts))
  (format t "~%")
  (format t "Starting benchmark for inserts~%")
  (with-timing (rts seconds)
      (run-workers workers retweets messageid *insert-rt* connspec)
    (format t "Inserting took ~f seconds, did ~d rts~%" seconds rts)))
(defun run-workers (workers retweets messageid sql
                    &optional (connspec *connspec*))
  (let* ((*kernel* (lparallel:make-kernel workers))
         (channel  (lparallel:make-channel)))
    (loop repeat workers
       do (lparallel:submit-task channel #'retweet-many-times
                                 retweets messageid sql connspec))
    (loop repeat workers sum (lparallel:receive-result channel))))
(defun retweet-many-times (times messageid sql
                           &optional (connspec *connspec*))
  (pomo:with-connection connspec
    (pomo:query
     (format nil "set application_name to 'worker ~a'"
             (lparallel:kernel-worker-index)))
    (loop repeat times sum (retweet messageid sql))))
(defun retweet (messageid sql)
  (handler-case
      (progn
        (pomo:query sql messageid)
        1)
    (database-error (c)
      (format t "Error: ~a~%" c)
      0)))

Performing Test

For the first test, it’s a typical ...