advanced/threads_wait_on_processes_and_sockets_no_parens.rb

tests/_examples/advanced/threads_wait_on_processes_and_sockets_no_parens.rb in the metorex repository.

# Each call below waits on a child process, a file or a socket that is ready
# only later, and a counter thread beside it shows the other threads ran
# while it waited.
require "socket"
require "tmpdir"

def counted name
  $count = 0
  counter = Thread.new { loop { $count += 1; Thread.pass } }
  Thread.pass
  $count = 0
  result = yield
  counter.kill
  counter.join
  puts "#{name}: #{$count > 0} #{result.inspect}"
end

counted("Kernel#sleep") { sleep(0.2).class }
counted("Kernel#system") { system("sleep 0.2") }
counted("Kernel#`") { `sleep 0.2; echo hi` }
counted("IO.popen") { IO.popen("sleep 0.2; echo hi") { |child| child.read } }
# The child reads nothing until a thread opens the gate, and that thread runs
# only once the write waits, so the pipe fills before the child drains it.
counted "IO.popen writing" do
  Dir.mktmpdir do |directory|
    gate = File.join directory, "gate"
    File.mkfifo gate
    IO.popen ["sh", "-c", "read go < \"$0\"; cat > /dev/null", gate], "w" do |child|
      Thread.new { File.write gate, "go\n" }
      child.write "x" * 200_000
    end
  end
end
counted("Process.wait") { Process.wait(spawn("sleep 0.2")).class }
counted("Process.wait2") { Process.wait2(spawn("sleep 0.2"))[1].success? }
counted("Process.waitpid") { Process.waitpid(spawn("sleep 0.2")).class }
counted("Process.waitpid2") { Process.waitpid2(spawn("sleep 0.2"))[1].success? }
counted("Process.waitall") { spawn("sleep 0.2"); Process.waitall.size }
counted("Process::Status.wait") { Process::Status.wait(spawn("sleep 0.2")).success? }

Dir.mktmpdir do |directory|
  locked = File.join(directory, "locked")
  File.write locked, ""
  holder = File.open(locked)
  holder.flock File::LOCK_EX
  Thread.new do
    sleep 0.2
    holder.flock File::LOCK_UN
  end
  counted("File#flock") { File.open(locked) { |waiting| waiting.flock(File::LOCK_EX) } }

  named = File.join(directory, "named")
  File.mkfifo named
  Thread.new do
    sleep 0.2
    File.write named, "hi\n"
  end
  counted("File.open on a FIFO") { File.open(named) { |fifo| fifo.read } }
  Thread.new do
    sleep 0.2
    File.write named, "hi\n"
  end
  counted("File.read on a FIFO") { File.read(named) }
  Thread.new do
    sleep 0.2
    File.read named
  end
  counted("File.write on a FIFO") { File.write(named, "hi\n") }

  # A FIFO opened while the other threads end partway, and one opened with
  # no other thread at all, are both read once the child writes.
  brief = Thread.new { 3.times { Thread.pass } }
  writer = spawn "sleep 0.2; echo partway > #{named}"
  puts File.open(named) { |fifo| fifo.read }
  Process.wait writer
  brief.join
  writer = spawn "sleep 0.2; echo alone > #{named}"
  puts File.open(named) { |fifo| fifo.read }
  Process.wait writer

  listening = UNIXServer.new(File.join(directory, "socket"))
  Thread.new do
    sleep 0.2
    UNIXSocket.new(listening.path).close
  end
  counted("UNIXServer#accept") { listening.accept.class }
end

server = TCPServer.new("127.0.0.1", 0)
port = server.addr[1]
Thread.new do
  sleep 0.2
  TCPSocket.new("127.0.0.1", port).close
end
counted("TCPServer#accept") { server.accept.class }

Thread.new do
  sleep 0.2
  connected = TCPSocket.new("127.0.0.1", port)
  sleep 0.2
  connected.write "hi"
  connected.close
end
counted("BasicSocket#recv") { server.accept.recv(10) }

Thread.new do
  sleep 0.2
  connected = TCPSocket.new("127.0.0.1", port)
  sleep 0.2
  connected.write "hi"
  connected.close
end
counted("BasicSocket#recvmsg") { server.accept.recvmsg(10)[0] }

datagrams = UDPSocket.new
datagrams.bind "127.0.0.1", 0
Thread.new do
  sleep 0.2
  UDPSocket.new.send("hi", 0, "127.0.0.1", datagrams.addr[1])
end
counted("UDPSocket#recvfrom") { datagrams.recvfrom(10)[0] }

plain = Socket.new(:INET, :STREAM)
plain.bind Addrinfo.tcp("127.0.0.1", 0)
plain.listen 1
Thread.new do
  sleep 0.2
  TCPSocket.new("127.0.0.1", plain.local_address.ip_port).close
end
counted("Socket#accept") { plain.accept[0].class }

# A host name is looked up while the other threads run.
busy = Thread.new { loop { Thread.pass } }
p Addrinfo.getaddrinfo("localhost", nil, :INET).map(&:ip_address).uniq
named_server = TCPServer.new "localhost", 0
named_port = named_server.addr[1]
p TCPSocket.new("localhost", named_port).class
p Socket.tcp("localhost", named_port).class
busy.kill

Output:

Kernel#sleep: true Integer
Kernel#system: true true
Kernel#`: true "hi\n"
IO.popen: true "hi\n"
IO.popen writing: true 200000
Process.wait: true Integer
Process.wait2: true true
Process.waitpid: true Integer
Process.waitpid2: true true
Process.waitall: true 1
Process::Status.wait: true true
File#flock: true 0
File.open on a FIFO: true "hi\n"
File.read on a FIFO: true "hi\n"
File.write on a FIFO: true 3
partway
alone
UNIXServer#accept: true UNIXSocket
TCPServer#accept: true TCPSocket
BasicSocket#recv: true "hi"
BasicSocket#recvmsg: true "hi"
UDPSocket#recvfrom: true "hi"
Socket#accept: true Socket
["127.0.0.1"]
TCPSocket
Socket