threading/channels.Chan is not free of data-races according to thread-sanitizers

[PhilippMDoerner]

Heyho, while working to clear up my own package of data-races I noticed that this example will declare a data-race in thread-sanitizer (tsan):

import threading/channels
type Thing = ref object

var chan = newChan[Thing]()
var thr: Thread[void]

proc loop() =
  while true:
    var msg: Thing
    let hasMsg = chan.tryRecv(msg)
    if hasMsg:
      break

proc main() =
  createThread(thr, loop)
  discard chan.trySend(Thing())

  joinThread(thr)
main()

Compiled with:
nim r --cc:clang --mm:orc -d:release -d:useMalloc -f --passc:"-fsanitize=thread -fno-omit-frame-pointer -mno-omit-leaf-frame-pointer" --debugger:native --passl:"-fsanitize=thread -fno-omit-frame-pointer -mno-omit-leaf-frame-pointer" src/playground.nim

tsan sees a data-race here between trySend with channelSend in one thread and tryRecv with channelReceive in the other. I'm not entirely sure how to interpret that, but I would assume it is not a false positive.

Even if the data-race does not pose a real problem, I would want to solve it because users of the lib (like me) would want to tsan-check their own applications to make them as stable as can be.

I'm missing too much of an understanding of threading/channels.Chan inner working to be able to tell if that is actually the case and what these two are racing through.

Edit: Beef stated he might've seen what caused the issue. I quote: "isFull is called before a lock happens". Would changing that be the solution here? Leorize suggested to use try-acquire to get that lock or using approaches without an if there, though I can't imagine them.

tsan stacktrace

```txt WARNING: ThreadSanitizer: data race (pid=2235220) Write of size 8 at 0x7b2c00000090 by main thread (mutexes: write M0): #0 channelSend__playground_u356 /home/philipp/dev/playground/channels.nim:192:130 (playground_1B88682F16D8AFE3768CDB5129F5ABAE744E02A7+0xed985) (BuildId: 92969305384c4f84ef7e6297746e94551b123ec1) #1 trySend__playground_u329 /home/philipp/.nimble/pkgs2/threading-0.2.0-3cd4360369b8abf1c53ddfd49ea8aef70208658c/threading/channels.nim:271:11 (playground_1B88682F16D8AFE3768CDB5129F5ABAE744E02A7+0xedf37) (BuildId: 92969305384c4f84ef7e6297746e94551b123ec1) #2 main__playground_u277 /home/philipp/dev/playground/src/playground.nim:16:27 (playground_1B88682F16D8AFE3768CDB5129F5ABAE744E02A7+0xedf37) #3 NimMainModule /home/philipp/dev/playground/src/playground.nim:19:2 (playground_1B88682F16D8AFE3768CDB5129F5ABAE744E02A7+0xee320) (BuildId: 92969305384c4f84ef7e6297746e94551b123ec1) #4 NimMainInner /home/philipp/dev/playground/src/playground.nim:41:2 (playground_1B88682F16D8AFE3768CDB5129F5ABAE744E02A7+0xee320) #5 NimMain /home/philipp/dev/playground/src/playground.nim:52:2 (playground_1B88682F16D8AFE3768CDB5129F5ABAE744E02A7+0xee320) #6 main /home/philipp/dev/playground/src/playground.nim:60:2 (playground_1B88682F16D8AFE3768CDB5129F5ABAE744E02A7+0xee320)

Previous read of size 8 at 0x7b2c00000090 by thread T1:
#0 channelReceive__playground_u179 /home/philipp/.nimble/pkgs2/threading-0.2.0-3cd4360369b8abf1c53ddfd49ea8aef70208658c/threading/channels.nim:205:22 (playground_1B88682F16D8AFE3768CDB5129F5ABAE744E02A7+0xed609) (BuildId: 92969305384c4f84ef7e6297746e94551b123ec1)
#1 tryRecv__playground_u167 /home/philipp/.nimble/pkgs2/threading-0.2.0-3cd4360369b8abf1c53ddfd49ea8aef70208658c/threading/channels.nim:281:11 (playground_1B88682F16D8AFE3768CDB5129F5ABAE744E02A7+0xedcab) (BuildId: 92969305384c4f84ef7e6297746e94551b123ec1)
#2 loop__playground_u165 /home/philipp/dev/playground/src/playground.nim:10:57 (playground_1B88682F16D8AFE3768CDB5129F5ABAE744E02A7+0xedcab)
#3 threadProcWrapDispatch__stdZtypedthreads_u105 /home/philipp/.choosenim/toolchains/nim-2.0.2/lib/system/threadimpl.nim:66:2 (playground_1B88682F16D8AFE3768CDB5129F5ABAE744E02A7+0xed059) (BuildId: 92969305384c4f84ef7e6297746e94551b123ec1)
#4 threadProcWrapStackFrame__stdZtypedthreads_u95 /home/philipp/.choosenim/toolchains/nim-2.0.2/lib/system/threadimpl.nim:95:2 (playground_1B88682F16D8AFE3768CDB5129F5ABAE744E02A7+0xed059)
#5 threadProcWrapper__stdZtypedthreads_u81 /home/philipp/.choosenim/toolchains/nim-2.0.2/lib/system/threadimpl.nim:101:2 (playground_1B88682F16D8AFE3768CDB5129F5ABAE744E02A7+0xe7b45) (BuildId: 92969305384c4f84ef7e6297746e94551b123ec1)

Location is heap block of size 176 at 0x7b2c00000000 allocated by main thread:
#0 malloc (playground_1B88682F16D8AFE3768CDB5129F5ABAE744E02A7+0x79b63) (BuildId: 92969305384c4f84ef7e6297746e94551b123ec1)
#1 allocChannel__OOZOOZOOZOnimbleZpkgs50Zthreading4548O50O484551cd52515448515457b56abf49c5351ddfd5257ea56aef5548504856545356cZthreadingZchannels_u40 /home/philipp/.nimble/pkgs2/threading-0.2.0-3cd4360369b8abf1c53ddfd49ea8aef70208658c/threading/channels.nim:134:24 (playground_1B88682F16D8AFE3768CDB5129F5ABAE744E02A7+0xed41c) (BuildId: 92969305384c4f84ef7e6297746e94551b123ec1)
#2 newChan__playground_u4 /home/philipp/.nimble/pkgs2/threading-0.2.0-3cd4360369b8abf1c53ddfd49ea8aef70208658c/threading/channels.nim:316:13 (playground_1B88682F16D8AFE3768CDB5129F5ABAE744E02A7+0xed54a) (BuildId: 92969305384c4f84ef7e6297746e94551b123ec1)
#3 NimMainModule /home/philipp/dev/playground/src/playground.nim:4:85 (playground_1B88682F16D8AFE3768CDB5129F5ABAE744E02A7+0xee2f6) (BuildId: 92969305384c4f84ef7e6297746e94551b123ec1)
#4 NimMainInner /home/philipp/dev/playground/src/playground.nim:41:2 (playground_1B88682F16D8AFE3768CDB5129F5ABAE744E02A7+0xee2f6)
#5 NimMain /home/philipp/dev/playground/src/playground.nim:52:2 (playground_1B88682F16D8AFE3768CDB5129F5ABAE744E02A7+0xee2f6)
#6 main /home/philipp/dev/playground/src/playground.nim:60:2 (playground_1B88682F16D8AFE3768CDB5129F5ABAE744E02A7+0xee2f6)

Mutex M0 (0x7b2c00000000) created at:
#0 pthread_mutex_init (playground_1B88682F16D8AFE3768CDB5129F5ABAE744E02A7+0x99f78) (BuildId: 92969305384c4f84ef7e6297746e94551b123ec1)
#1 initLock__coreZlocks_u7 /home/philipp/.choosenim/toolchains/nim-2.0.2/lib/core/locks.nim:38:2 (playground_1B88682F16D8AFE3768CDB5129F5ABAE744E02A7+0xed45e) (BuildId: 92969305384c4f84ef7e6297746e94551b123ec1)
#2 allocChannel__OOZOOZOOZOnimbleZpkgs50Zthreading4548O50O484551cd52515448515457b56abf49c5351ddfd5257ea56aef5548504856545356cZthreadingZchannels_u40 /home/philipp/.nimble/pkgs2/threading-0.2.0-3cd4360369b8abf1c53ddfd49ea8aef70208658c/threading/channels.nim:139:2 (playground_1B88682F16D8AFE3768CDB5129F5ABAE744E02A7+0xed45e)
#3 newChan__playground_u4 /home/philipp/.nimble/pkgs2/threading-0.2.0-3cd4360369b8abf1c53ddfd49ea8aef70208658c/threading/channels.nim:316:13 (playground_1B88682F16D8AFE3768CDB5129F5ABAE744E02A7+0xed54a) (BuildId: 92969305384c4f84ef7e6297746e94551b123ec1)
#4 NimMainModule /home/philipp/dev/playground/src/playground.nim:4:85 (playground_1B88682F16D8AFE3768CDB5129F5ABAE744E02A7+0xee2f6) (BuildId: 92969305384c4f84ef7e6297746e94551b123ec1)
#5 NimMainInner /home/philipp/dev/playground/src/playground.nim:41:2 (playground_1B88682F16D8AFE3768CDB5129F5ABAE744E02A7+0xee2f6)
#6 NimMain /home/philipp/dev/playground/src/playground.nim:52:2 (playground_1B88682F16D8AFE3768CDB5129F5ABAE744E02A7+0xee2f6)
#7 main /home/philipp/dev/playground/src/playground.nim:60:2 (playground_1B88682F16D8AFE3768CDB5129F5ABAE744E02A7+0xee2f6)

Thread T1 (tid=2235222, running) created by main thread at:
#0 pthread_create (playground_1B88682F16D8AFE3768CDB5129F5ABAE744E02A7+0x64a36) (BuildId: 92969305384c4f84ef7e6297746e94551b123ec1)
#1 createThread__stdZtypedthreads_u60 /home/philipp/.choosenim/toolchains/nim-2.0.2/lib/std/typedthreads.nim:246:103 (playground_1B88682F16D8AFE3768CDB5129F5ABAE744E02A7+0xe7c79) (BuildId: 92969305384c4f84ef7e6297746e94551b123ec1)
#2 createThread__stdZtypedthreads_u51 /home/philipp/.choosenim/toolchains/nim-2.0.2/lib/std/typedthreads.nim:262:2 (playground_1B88682F16D8AFE3768CDB5129F5ABAE744E02A7+0xe7d70) (BuildId: 92969305384c4f84ef7e6297746e94551b123ec1)
#3 main__playground_u277 /home/philipp/dev/playground/src/playground.nim:15:2 (playground_1B88682F16D8AFE3768CDB5129F5ABAE744E02A7+0xede8a) (BuildId: 92969305384c4f84ef7e6297746e94551b123ec1)
#4 NimMainModule /home/philipp/dev/playground/src/playground.nim:19:2 (playground_1B88682F16D8AFE3768CDB5129F5ABAE744E02A7+0xee320) (BuildId: 92969305384c4f84ef7e6297746e94551b123ec1)
#5 NimMainInner /home/philipp/dev/playground/src/playground.nim:41:2 (playground_1B88682F16D8AFE3768CDB5129F5ABAE744E02A7+0xee320)
#6 NimMain /home/philipp/dev/playground/src/playground.nim:52:2 (playground_1B88682F16D8AFE3768CDB5129F5ABAE744E02A7+0xee320)
#7 main /home/philipp/dev/playground/src/playground.nim:60:2 (playground_1B88682F16D8AFE3768CDB5129F5ABAE744E02A7+0xee320)

SUMMARY: ThreadSanitizer: data race /home/philipp/dev/playground/channels.nim:192:130 in channelSend__playground_u356

ThreadSanitizer: reported 1 warnings

<details>

[ZoomRmc]

Here's what's happening in the try paths of the send/receive procs (removing all the static branching):

if chan.isFull(): return false
acquire(chan.lock)
if chan.isFull():
  release(chan.lock)
  return false

My understanding of the logic I've got when I was refactoring the code earlier is that the first isFull check is a first-line of defense in preventing lock contention. Without it, in cases like the code in the OP, the threads would try to acquire the lock extremely fast and then do nothing and release it, thus burning the resources. Even if the race happens for the first isFull/isEmpty check (that is never guaranteed to return an actual state of the channel, unless you're inside the lock's critical section), it shouldn't result in any incorrect behaviour since the state is checked again (now holding the lock) before acting on the buffer.

So, if I'm not forgetting some edge case it's supposed to work for, this is an optimization and It would be beneficial to see if it really achieves any improved performance in real world cases (or at least tests). It's also interesting to see what tsan has to say about the original C code where basically the same order of operations happens.

I don't know if tryAcquire would help in this case, as it changes the contract. The procs will start bailing out on locking failures (in addition to no-space/no-data occurrences) making code even less deterministic, which is the opposite of what we want.

[PhilippMDoerner]

Hmmm I'll look deeper into this to philosophize about it.
I noticed very similar behaviour in the system.Channel implementation - is this possibly the same issue there (as they might share optimization ideas there) ?

[ZoomRmc]

BTW, I understand the code is just a minimal reproducing snippet, but just in case, for others reading: try to avoid calling trySend/tryRecv in a loop, the overhead is probably enormous if the body is not busy enough.

[PhilippMDoerner]

For reference for any curious future reader, I wanted to have some kind of data to back up the assertion that this if-check was valid. So with some feedback from Zoom I wrote an example that creates my own versions of trySafe and tryRecv, one with, the other without this extra if-check. I also added in incrementing a counter to check for when an acquire happens that would've otherwise been prevented by the "unsafe" version of the access.

So what this does is spawn N producer threads and M consumer threads. Producer threads just collectively try to send 1000 messages through a channel of size X (either 1,5,10,50,100 or 1000), while consumer threads try to read and atomically increment a counter.

This is done 1000 times per configuration of threads and channel sizes and that "microbenchmark" is repeated once for the "Optimized" version (containing the data-race) and once for the "Safe" version. It spits the result out as CSV, basically how long sending those 1000 messages took on average.

The "benchmark" code... way too long

import threading/channels {.all.}
import std/[locks, os, math, algorithm, strutils, strformat, sequtils, isolation, atomics, times]

# === SETUP START === #
var sendLockAcquireInst: Atomic[int]
proc customChannelSend(chan: ChannelRaw, data: pointer, size: int, blocking: static bool, optimize: static bool): bool =
  assert not chan.isNil
  assert not data.isNil

  when optimize:
    if chan.isFull(): return false

  acquire(chan.lock)

  # check for when another thread was faster to fill
  when blocking:
    while chan.isFull():
      wait(chan.spaceAvailableCV, chan.lock)
  else:
    if chan.isFull():
      sendLockAcquireInst.atomicInc
      release(chan.lock)
      return false

  assert not chan.isFull()

  let writeIdx = if chan.head < chan.slots:
      chan.head
    else:
      chan.head - chan.slots

  copyMem(chan.buffer[writeIdx * size].addr, data, size)

  inc(chan.head)
  if chan.head == 2 * chan.slots:
    chan.head = 0

  signal(chan.dataAvailableCV)
  release(chan.lock)
  result = true

proc trySendSafe*[T](c: Chan[T], src: sink Isolated[T]): bool {.inline.} =
  ## Sends item to the channel (non-blocking). Version not containing optimization, not containing data-race.
  var data = src.extract
  result = customChannelSend(c.d, data.unsafeAddr, sizeof(T), false, false)
  if result:
    wasMoved(data)

proc trySendOpt*[T](c: Chan[T], src: sink Isolated[T]): bool {.inline.} =
  ## Sends item to the channel (non-blocking). Version not containing optimization, not containing data-race.
  var data = src.extract
  result = customChannelSend(c.d, data.unsafeAddr, sizeof(T), false, true)
  if result:
    wasMoved(data)

var receiveLockAcquireInst: Atomic[int]
proc customChannelReceive(chan: ChannelRaw, data: pointer, size: int, blocking: static bool, optimize: static bool): bool =
  assert not chan.isNil
  assert not data.isNil

  when optimize:
    if chan.isEmpty(): return false

  acquire(chan.lock)

  # check for when another thread was faster to empty
  when blocking:
    while chan.isEmpty():
      wait(chan.dataAvailableCV, chan.lock)
  else:
    if chan.isEmpty():
      receiveLockAcquireInst.atomicInc
      release(chan.lock)
      return false

  assert not chan.isEmpty()

  let readIdx = if chan.tail < chan.slots:
      chan.tail
    else:
      chan.tail - chan.slots

  copyMem(data, chan.buffer[readIdx * size].addr, size)

  inc(chan.tail)
  if chan.tail == 2 * chan.slots:
    chan.tail = 0

  signal(chan.spaceAvailableCV)
  release(chan.lock)
  result = true

proc tryRecvSafe*[T](c: Chan[T], dst: var T): bool {.inline.} =
  ## Receives item from the channel (non-blocking). Optimized version
  customChannelReceive(c.d, dst.addr, sizeof(T), false, false)

proc tryRecvOpt*[T](c: Chan[T], dst: var T): bool {.inline.} =
  ## Receives item from the channel (non-blocking). Optimized version
  customChannelReceive(c.d, dst.addr, sizeof(T), false, true)

# === SETUP END === #
type Thing = ref object
  txt: string

const MESSAGE_COUNT = 1_000
const ITERATION_COUNT = 1_000
const MAX_THREAD_COUNT = 5
const CHANNEL_SIZES = @[
  1,
  5,
  10,
  50,
  100,
  1_000,
]

var IS_RUNNING = true
let msgs = (1..MESSAGE_COUNT).mapIt(Thing(txt: "#" & $it))

type ThreadData = object
  channelIndex: 0..CHANNEL_SIZES.len()
  threadIndex: 0..4
  threadCount: 1..5

var consumerThreads: seq[Thread[ThreadData]] = (1..MAX_THREAD_COUNT).mapIt(Thread[ThreadData]())
var producerThreads: seq[Thread[ThreadData]] = (1..MAX_THREAD_COUNT).mapIt(Thread[ThreadData]())


var optChannels = CHANNEL_SIZES.mapIt(newChan[Thing](it))
var optCounter: Atomic[int]
proc optLoop(data: ThreadData) {.thread, nimcall.} =
  {.gcsafe.}:
    while optCounter.load() < MESSAGE_COUNT:
      var msg: Thing
      let hasMsg = optChannels[data.channelIndex].tryRecvOpt(msg)
      if hasMsg:
        optCounter.atomicInc
      

var safeChannels = CHANNEL_SIZES.mapIt(newChan[Thing](it))
var safeCounter: Atomic[int]
proc safeLoop(data: ThreadData) {.thread , nimcall.} =
  {.gcsafe.}:
    while safeCounter.load() < MESSAGE_COUNT:
      var msg: Thing
      let hasMsg = safeChannels[data.channelIndex].tryRecvSafe(msg)
      if hasMsg:
        safeCounter.atomicInc

proc clientOptLoop(data: ThreadData) {.thread, nimcall.} =
  {.gcsafe.}:
    for index, msg in msgs:
      if index.mod(data.threadCount) != data.threadIndex:
        continue
      
      while not optChannels[data.channelIndex].trySendOpt(unsafeIsolate(msg)): 
        discard

proc clientSafeLoop(data: ThreadData) {.thread, nimcall.} =
  {.gcsafe.}:
    for index, msg in msgs:
      if index.mod(data.threadCount) != data.threadIndex:
        continue
      
      while not safeChannels[data.channelIndex].trySendSafe(unsafeIsolate(msg)): 
        discard

proc clean() =
  optCounter.store(0)
  safeCounter.store(0)
  receiveLockAcquireInst.store(0)
  sendLockAcquireInst.store(0)
  for chan in optChannels:
    assert chan.peek == 0
  
  for chan in safeChannels:
    assert chan.peek == 0

template benchBlock(name: string; iterations, consumerThreadCount, producerThreadCount: int; callBlock: untyped) =
  var benchtimes: seq[float] = @[] 
  for index in 0..iterations:
    clean()
    var t0 = cpuTime()

    callBlock
    
    var t1 = cpuTime()
    benchtimes.add((t1-t0)*1000)
  
  benchtimes.sort()
  let totalIterationCount = MESSAGE_COUNT * iterations
  let medianIndex: int = int(benchtimes.high/2)
  let median: float = benchtimes[medianIndex]
  let max: float = benchtimes[^1]
  let min: float = benchtimes[1]
  let avg: float = benchtimes.sum() / iterations
  let channelSize: int = CHANNEL_SIZES[index]
  
  let columns = @[
    name,
    $consumerThreadCount,
    $producerThreadCount,
    $channelSize,
    $sendLockAcquireInst.load(),
    $receiveLockAcquireInst.load(),
    $median,
    $avg,
    $max,
    $min
  ].join(",")
  echo columns
  
proc main() =
  let columns = @[
    "Type",
    "ConsumerThreadCount",
    "ProducerThreadCount",
    "ChannelSize",
    "Bad Send Acquires",
    "Bad Recv Acquires",
    "Median (µs)",
    "Avg (µs)",
    "Max (µs)",
    "Min (µs)"
  ].join(",")
  echo "MessageCountPerIteration,", MESSAGE_COUNT, ",TotalMessageCount,", MESSAGE_COUNT * ITERATION_COUNT, "\n"
  echo columns
  for consumerThreadCount in 1..consumerThreads.high:
    for producerThreadCount in 1..producerThreads.high:
      for index in 0..CHANNEL_SIZES.high:
        benchBlock("Optimized", ITERATION_COUNT, consumerThreadCount, producerThreadCount):
          for threadIndex in 0..(consumerThreadCount - 1):
            createThread(consumerThreads[threadIndex], optloop, ThreadData(channelIndex: index, threadIndex: threadIndex, threadCount: consumerThreadCount))
          
          for threadIndex in 0..(producerThreadCount - 1):
            createThread(producerThreads[threadIndex], clientOptLoop, ThreadData(channelIndex: index, threadIndex: threadIndex, threadCount: producerThreadCount))
        
          while optCounter.load() < MESSAGE_COUNT: sleep(0)
          
          for threadIndex in 0..(producerThreadCount - 1):
            joinThread(producerThreads[threadIndex])
          
          for threadIndex in 0..(consumerThreadCount - 1):
            joinThread(consumerThreads[threadIndex])
        
        benchBlock("Safe", ITERATION_COUNT, consumerThreadCount, producerThreadCount):
          for threadIndex in 0..(consumerThreadCount - 1):
            createThread(consumerThreads[threadIndex], safeloop, ThreadData(channelIndex: index, threadIndex: threadIndex, threadCount: consumerThreadCount))
          
          for threadIndex in 0..(producerThreadCount - 1):
            createThread(producerThreads[threadIndex], clientSafeLoop, ThreadData(channelIndex: index, threadIndex: threadIndex, threadCount: producerThreadCount))

          while safeCounter.load() < MESSAGE_COUNT: sleep(0)

          for threadIndex in 0..(producerThreadCount - 1):
            joinThread(producerThreads[threadIndex])
          
          for threadIndex in 0..(consumerThreadCount - 1):
            joinThread(consumerThreads[threadIndex])
        
main()

echo "Done"

benchy.csv
benchy.ods

In general this is a limited benchmark, since it includes spawning and joining the thread inside the benchmark.
However, it does show that there's a fair amount of lock-acquisitions that get prevented by the data-racey "isFull"/"isEmpty" checks. Around 98% of "pointless" lock acquisitions get removed this way, particularly when lock acquisition is fairly contentious by having multiple producers/consumers on one very small channel.

[mratsim]

All shared variable accesses should be protected by atomics or locks or ASAN will complain.

head and tail can be atomics, and use load(moRelaxed) and ASAN will not complain anymore.

[PhilippMDoerner]

TSAN is the particular thing complaining here ;-)

But I agree, that could work nicely. I'd be happy to write up a PR on this one if @ZoomRmc isn't already on it.

Either way I think it might make sense to introduce general asan/tsan checks in general, just to be able to provide hard guarantees like "This has no data-races and does not leak!".

[PhilippMDoerner]

@mratsim Could you take a look at zooms comment in #56 (comment)
Not entirely sure what the response here should be.