Change thread status while waiting data from effil.channel

[effil]

Change thread status while waiting data from effil.channel

[mihacooper]

Hmm, the usability of this feature is not clear for me. Can you provide some real life example of usage?

[devildevilson]

I cannot effectively check that all of the threads have finished their work and now waiting for the new data from the channel. This can ensure the programmer that all concurrent data have the final state between nominal program steps. Example:

local channel = effil.channel()
local free_threads = effil.channel()
function consumer(channel, free_threads) 
  while true do
    local func = channel:pop()
    free_threads:pop() -- channel:pop() and free_threads:pop() dont perform atomically
    func()
    free_threads:push(1)
  end
end

local many_threads = 10
local threads = {}
for i = 0, many_threads-1 do
  threads[i] = effil.thread(consumer)(channel, free_threads)
  free_threads:push(1)
end

function wait()
  repeat
    -- waiting
  until free_threads:size() == many_threads and channel:size() == 0 -- can be false positive

  --[[
  repeat
    local counter = 0
    for i = 0, many_threads-1 do
      local status, err, trace = threads[i]:status()
      if status == "waiting" then counter = counter + 1 end
    end
  until counter == many_threads and channel:size() == 0 -- new wait() doesnt produce false positives
  --]]
end

function job() 
  effil.sleep(2)
end

-- nominal program step 1, lets assume that job() is changing state in shared table
local jobs = 50 
for i = 0, jobs do
  channel:push(job())
end

wait() -- this function can get us many false positives
-- but we need to ensure that any data in shared table is in it final state

-- nominal program step 2, lets assume that job() is reading random data from shared table
local jobs2 = 50 
for i = 0, jobs2 do
  channel:push(job())
end

[mihacooper]

Dedicated thread's status can't give you required guarantee. Cause right after it becomes in waiting status someone can put message in channel and thread will stop waiting.

So, speaking about your task - waiting until all works is done. You rather need functionality for waiting on multiple primitives. Using that you'll be able to do something like:

effil.wait_for_all(thread1_finished, thread2_finished, ...)

Where thread1_finished can be one of: conditional variable or channel (with capacity = 1).
Both of these features (waiting on multiple primitives and conditional variables) were in our plans, but they are not foremost right now.

[devildevilson]

My bad, the consumer() function I provide is wrong. The channel:pop() must be in never ending loop (see below). Thus I doubt that effil.wait_for_all can gave me expected behavior, cause threads never get status "completed" in this case. But the example may benefit from atomically changing status of thread, when it is waiting for the new data in channel. In this case I can simply count waiting threads, and if them == threads_count and channel is empty, that is the signal to program to go to the next step.

New consumer (I'll update the previous message):

function consumer(channel, free_threads)
  while true do
    local func = channel:pop()
    free_threads:pop()
    func()
    free_threads:push(1)
  end
end

New wait func:

function wait()
  repeat
    local counter = 0
    for i = 0, many_threads-1 do
      local status, err, trace = threads[i]:status()
      if status == "waiting" then counter = counter + 1 end
    end
  until counter == many_threads and channel:size() == 0
end

[mihacooper]

1. I have told to use effil.wait_for_all for waiting on some additional primitive, not on the thread object. So, your note about threads never get status "completed" is not relevant.
2. Your wait function will not work correct in case someone put message in channel (during wait execution). If it's not a problem then you can just check the size of free_threads queue size it gives the same result as special waiting status for threads.

My point is if you need to synchronously check/wait for states of multiple objects (threads, channels, etc) you have to use low-level functionality like effil.wait_for_all or use some special counters locked by mutex.

[devildevilson]

Understood. Thank you for the feedback. Looking forward to the effil.wait_for_all.