класс Thread
Потоки — это реализация Ruby модели параллельного программирования.
Программы, которым требуется несколько потоков выполнения, отлично подходят для класса Ruby Thread.
Например, мы можем создать новый поток, отдельный от выполнения основного потока, с помощью ::new.
thr = Thread.new { puts "What's the big deal" }
Затем мы можем приостановить выполнение основного потока и позволить новому потоку завершиться с помощью join:
thr.join #=> "What's the big deal"
Если мы не вызовем thr.join до завершения основного потока, все остальные потоки, включая thr, будут уничтожены.
В качестве альтернативы для управления несколькими потоками одновременно можно использовать массив, как в следующем примере:
threads = []
threads << Thread.new { puts "What's the big deal" }
threads << Thread.new { 3.times { puts "Threads are fun!" } }
Создав несколько потоков, мы ждём их последовательного завершения.
threads.each { |thr| thr.join }
Чтобы получить последнее значение потока, используйте value
thr = Thread.new { sleep 1; "Useful value" }
thr.value #=> "Useful value"
Thread — инициализация
Для создания новых потоков Ruby предоставляет методы ::new, ::start и ::fork. Для каждого из этих методов необходимо передать блок, иначе будет вызвано исключение ThreadError.
При создании подкласса класса Thread метод initialize этого подкласса будет проигнорирован методами ::start и ::fork. В противном случае не забудьте вызвать super в методе initialize.
Thread — завершение
Ruby предоставляет несколько способов завершения потоков.
Метод класса ::kill предназначен для завершения заданного потока:
thr = Thread.new { sleep }
Thread.kill(thr) # sends exit() to thr
В качестве альтернативы можно использовать метод экземпляра exit или его псевдонимы kill и terminate.
thr.exit
Thread — состояние
Ruby предоставляет несколько методов экземпляра для получения сведений о состоянии потока. Чтобы получить строку с текущим состоянием потока, используйте status
thr = Thread.new { sleep }
thr.status # => "sleep"
thr.exit
thr.status # => false
Также можно использовать alive?, чтобы узнать, выполняется поток или находится в состоянии ожидания, и stop?, чтобы узнать, завершён поток или находится в состоянии ожидания.
Thread — переменные и область видимости
Поскольку потоки создаются с помощью блоков, для области видимости переменных действуют те же правила, что и для других блоков Ruby. Любые локальные переменные, созданные внутри этого блока, доступны только этому потоку.
Локальные переменные волокна и локальные переменные потока
У каждого волокна есть собственное хранилище для Thread#[]. Новая локальная переменная волокна доступна только внутри этого Fiber. Например:
Thread.new {
Thread.current[:foo] = "bar"
Fiber.new {
p Thread.current[:foo] # => nil
}.resume
}.join
В этом примере для получения локальных переменных волокна используется [], а для задания — []=. Также можно использовать keys, чтобы вывести список локальных переменных волокна для заданного потока, и key?, чтобы проверить, существует ли такая переменная.
Локальные переменные потока доступны во всей области видимости потока. Рассмотрим следующий пример:
Thread.new{
Thread.current.thread_variable_set(:foo, 1)
p Thread.current.thread_variable_get(:foo) # => 1
Fiber.new{
Thread.current.thread_variable_set(:foo, 2)
p Thread.current.thread_variable_get(:foo) # => 2
}.resume
p Thread.current.thread_variable_get(:foo) # => 2
}.join
Видно, что локальная переменная потока :foo передалась волокну и к завершению потока получила значение 2.
В этом примере для создания новых локальных переменных потока используется thread_variable_set, а для обращения к ним — thread_variable_get.
Также доступны thread_variables для вывода списка всех локальных переменных потока и thread_variable? для проверки существования заданной локальной переменной потока.
Exception — обработка исключений
Если внутри потока возникает необработанное исключение, поток завершится. По умолчанию это исключение не будет передано другим потокам. Исключение сохраняется, и когда другой поток вызывает value или join, оно будет повторно вызвано в этом потоке.
t = Thread.new{ raise 'something went wrong' }
t.value #=> RuntimeError: something went wrong
Исключение можно вызвать из другого потока с помощью метода экземпляра Thread#raise, который принимает те же параметры, что и Kernel#raise.
Если задать Thread.abort_on_exception = true, Thread#abort_on_exception = true или $DEBUG = true, последующее необработанное исключение в потоке будет автоматически повторно вызвано в основном потоке.
С появлением метода класса ::handle_interrupt стало возможно асинхронно обрабатывать исключения в потоках.
Планирование
Ruby предоставляет несколько способов управления планированием потоков в программе.
Первый способ — использовать метод класса ::stop, чтобы приостановить текущий поток и запланировать выполнение другого потока.
Когда поток находится в состоянии ожидания, можно использовать метод экземпляра wakeup, чтобы сделать его доступным для планирования.
Также можно попробовать вызвать ::pass, который пытается передать выполнение другому потоку. Переключится ли выполняющийся поток, зависит от операционной системы. То же относится к методу priority, который позволяет указать планировщику потоков, каким потокам следует отдавать приоритет при передаче выполнения. Этот метод также зависит от операционной системы и на некоторых платформах может игнорироваться.
Общедоступные методы класса
static VALUE
rb_thread_s_abort_exc(VALUE _)
{
return RBOOL(GET_THREAD()->vm->thread_abort_on_exception);
} Возвращает состояние глобального параметра «прерывать выполнение при исключении».
По умолчанию задано значение false.
Если задано значение true, то при прерывании любого потока исключением это исключение будет повторно вызвано в основном потоке.
Также можно задать с помощью глобального флага $DEBUG или параметра командной строки -d.
См. также ::abort_on_exception=.
Для задания этого параметра отдельному потоку существует также метод уровня экземпляра; см. abort_on_exception.
static VALUE
rb_thread_s_abort_exc_set(VALUE self, VALUE val)
{
GET_THREAD()->vm->thread_abort_on_exception = RTEST(val);
return val;
} Если задано значение true, то при прерывании любого потока исключением это исключение будет повторно вызвано в основном потоке. Возвращает новое состояние.
Thread.abort_on_exception = true t1 = Thread.new do puts "In new thread" raise "Exception from thread" end sleep(1) puts "not reached"
В результате будет выведено:
In new thread prog.rb:4: Exception from thread (RuntimeError) from prog.rb:2:in `initialize' from prog.rb:2:in `new' from prog.rb:2
См. также ::abort_on_exception.
Для задания этого параметра отдельному потоку существует также метод уровня экземпляра; см. abort_on_exception=.
static VALUE
thread_s_current(VALUE klass)
{
return rb_thread_current();
} Возвращает выполняющийся в данный момент поток.
Thread.current #=> #<Thread:0x401bdf4c run>
static VALUE
each_caller_location(int argc, VALUE *argv, VALUE _)
{
rb_execution_context_t *ec = GET_EC();
long n, lev = ec_backtrace_range(ec, argc, argv, 1, 1, &n);
if (lev >= 0 && n != 0) {
rb_ec_partial_backtrace_object(ec, lev, n, NULL, FALSE, TRUE);
}
return Qnil;
} Передаёт в блок каждый кадр текущего стека выполнения в виде объекта расположения в трассировке стека.
static VALUE
thread_start(VALUE klass, VALUE args)
{
struct thread_create_params params = {
.type = thread_invoke_type_proc,
.args = args,
.proc = rb_block_proc(),
};
return thread_create_core(rb_thread_alloc(klass), ¶ms);
} По сути, то же, что и ::new. Однако если класс Thread является суперклассом, вызов start в этом подклассе не вызовет метод initialize подкласса.
static VALUE
rb_thread_s_handle_interrupt(VALUE self, VALUE mask_arg)
{
VALUE mask = Qundef;
rb_execution_context_t * volatile ec = GET_EC();
rb_thread_t * volatile th = rb_ec_thread_ptr(ec);
volatile VALUE r = Qnil;
enum ruby_tag_type state;
if (!rb_block_given_p()) {
rb_raise(rb_eArgError, "block is needed.");
}
mask_arg = rb_to_hash_type(mask_arg);
if (OBJ_FROZEN(mask_arg) && rb_hash_compare_by_id_p(mask_arg)) {
mask = Qnil;
}
rb_hash_foreach(mask_arg, handle_interrupt_arg_check_i, (VALUE)&mask);
if (UNDEF_P(mask)) {
return rb_yield(Qnil);
}
if (!RTEST(mask)) {
mask = mask_arg;
}
else if (RB_TYPE_P(mask, T_HASH)) {
OBJ_FREEZE(mask);
}
rb_ary_push(th->pending_interrupt_mask_stack, mask);
if (!rb_threadptr_pending_interrupt_empty_p(th)) {
th->pending_interrupt_queue_checked = 0;
RUBY_VM_SET_INTERRUPT(th->ec);
}
EC_PUSH_TAG(th->ec);
if ((state = EC_EXEC_TAG()) == TAG_NONE) {
r = rb_yield(Qnil);
}
EC_POP_TAG();
rb_ary_pop(th->pending_interrupt_mask_stack);
if (!rb_threadptr_pending_interrupt_empty_p(th)) {
th->pending_interrupt_queue_checked = 0;
RUBY_VM_SET_INTERRUPT(th->ec);
}
RUBY_VM_CHECK_INTS(th->ec);
if (state) {
EC_JUMP_TAG(th->ec, state);
}
return r;
} Изменяет момент обработки асинхронных прерываний.
Прерывание — это асинхронное событие и соответствующая процедура, вызванная методами Thread#raise, Thread#kill, обработчиком сигнала (пока не поддерживается) или завершением основного потока (если основной поток завершается, все остальные потоки будут уничтожены).
Переданный hash содержит пары вида ExceptionClass => :TimingSymbol. ExceptionClass — это класс исключения, обрабатываемого заданным блоком. TimingSymbol может быть одним из следующих символов:
:immediate-
Вызывать прерывания немедленно.
:on_blocking-
Вызывать прерывания во время BlockingOperation.
:never-
Никогда не вызывать прерывания.
BlockingOperation — это операция, блокирующая вызывающий поток, например чтение или запись. В реализации CRuby BlockingOperation — это любая операция, выполняемая без GVL.
Обработка замаскированных асинхронных прерываний откладывается до тех пор, пока они не будут разрешены. Этот метод аналогичен sigprocmask(3).
ПРИМЕЧАНИЕ
Асинхронные прерывания сложно использовать.
Для обмена данными между потоками рассмотрите альтернативный способ, например Queue.
Или используйте их, только если хорошо понимаете работу этого метода.
Использование
В этом примере мы защищаемся от исключений, вызванных Thread#raise.
При использовании символа TimingSymbol :never исключение RuntimeError всегда будет игнорироваться в первом блоке основного потока. Во втором блоке ::handle_interrupt мы можем намеренно обрабатывать исключения RuntimeError.
th = Thread.new do
Thread.handle_interrupt(RuntimeError => :never) {
begin
# You can write resource allocation code safely.
Thread.handle_interrupt(RuntimeError => :immediate) {
# ...
}
ensure
# You can write resource deallocation code safely.
end
}
end
Thread.pass
# ...
th.raise "stop"
Пока мы игнорируем исключение RuntimeError, можно безопасно выполнять код выделения ресурсов. Затем в блоке ensure можно безопасно освободить эти ресурсы.
Настройки управления стеком
Можно вкладывать несколько уровней блоков ::handle_interrupt, чтобы одновременно управлять несколькими классами ExceptionClass и символами TimingSymbol.
Thread.handle_interrupt(FooError => :never) {
Thread.handle_interrupt(BarError => :never) {
# FooError and BarError are prohibited.
}
}
Наследование для ExceptionClass
Учитываются все исключения, унаследованные от параметра ExceptionClass.
Thread.handle_interrupt(Exception => :never) {
# all exceptions inherited from Exception are prohibited.
}
Для обработки всех прерываний используйте Object, а не Exception в качестве ExceptionClass, поскольку прерывания kill/terminate не обрабатываются классом Exception.
static VALUE
rb_thread_s_ignore_deadlock(VALUE _)
{
return RBOOL(GET_THREAD()->vm->thread_ignore_deadlock);
} Возвращает состояние глобального параметра «игнорировать взаимоблокировку». По умолчанию задано значение false, поэтому взаимоблокировки не игнорируются.
См. также ::ignore_deadlock=.
static VALUE
rb_thread_s_ignore_deadlock_set(VALUE self, VALUE val)
{
GET_THREAD()->vm->thread_ignore_deadlock = RTEST(val);
return val;
} Возвращает новое состояние. Если задано значение true, виртуальная машина не будет проверять наличие взаимоблокировок. Этот параметр полезен только в том случае, если приложение может устранить взаимоблокировку другим способом, например с помощью сигнала.
Thread.ignore_deadlock = true
queue = Thread::Queue.new
trap(:SIGUSR1){queue.push "Received signal"}
# raises fatal error unless ignoring deadlock
puts queue.pop
См. также ::ignore_deadlock.
static VALUE
rb_thread_s_kill(VALUE obj, VALUE th)
{
return rb_thread_kill(th);
} Заставляет заданный thread завершиться; см. также Thread::exit.
count = 0
a = Thread.new { loop { count += 1 } }
sleep(0.1) #=> 0
Thread.kill(a) #=> #<Thread:0x401b3d30 dead>
count #=> 93947
a.alive? #=> false
static VALUE
thread_list(VALUE _)
{
return rb_thread_list();
} Возвращает массив объектов Thread для всех потоков, которые могут выполняться или остановлены.
Thread.new { sleep(200) }
Thread.new { 1000000.times {|i| i*i } }
Thread.new { Thread.stop }
Thread.list.each {|t| p t}
В результате будет выведено:
#<Thread:0x401b3e84 sleep> #<Thread:0x401b3f38 run> #<Thread:0x401b3fb0 sleep> #<Thread:0x401bdf4c run>
static VALUE
rb_thread_s_main(VALUE klass)
{
return rb_thread_main();
} Возвращает основной поток.
static VALUE
thread_s_new(int argc, VALUE *argv, VALUE klass)
{
rb_thread_t *th;
VALUE thread = rb_thread_alloc(klass);
if (GET_RACTOR()->threads.main->status == THREAD_KILLED) {
rb_raise(rb_eThreadError, "can't alloc thread");
}
rb_obj_call_init_kw(thread, argc, argv, RB_PASS_CALLED_KEYWORDS);
th = rb_thread_ptr(thread);
if (!threadptr_initialized(th)) {
rb_raise(rb_eThreadError, "uninitialized thread - check '%"PRIsVALUE"#initialize'",
klass);
}
return thread;
} Создаёт новый поток, выполняющий заданный блок.
Все args, переданные в ::new, будут переданы блоку:
arr = []
a, b, c = 1, 2, 3
Thread.new(a,b,c) { |d,e,f| arr << d << e << f }.join
arr #=> [1, 2, 3]
Если вызвать ::new без блока, будет вызвано исключение ThreadError.
Если вы создаёте подкласс Thread, не забудьте вызвать super в методе initialize, иначе будет вызвано исключение ThreadError.
static VALUE
thread_s_pass(VALUE klass)
{
rb_thread_schedule();
return Qnil;
} Передаёт планировщику потоков подсказку о передаче выполнения другому потоку. Выполняющийся поток может переключиться, а может и нет — это зависит от операционной системы и процессора.
static VALUE
rb_thread_s_pending_interrupt_p(int argc, VALUE *argv, VALUE self)
{
return rb_thread_pending_interrupt_p(argc, argv, GET_THREAD()->self);
} Возвращает значение, указывающее, пуста ли асинхронная очередь.
Поскольку с помощью Thread::handle_interrupt можно откладывать асинхронные события, этот метод позволяет определить, есть ли отложенные события.
Если этот метод возвращает true, можно завершить блоки :never.
Например, следующий метод немедленно обрабатывает отложенные асинхронные события.
def Thread.kick_interrupt_immediately
Thread.handle_interrupt(Object => :immediate) {
Thread.pass
}
end
Если задано error, проверяются только отложенные события типа error.
Использование
th = Thread.new{
Thread.handle_interrupt(RuntimeError => :on_blocking){
while true
...
# reach safe point to invoke interrupt
if Thread.pending_interrupt?
Thread.handle_interrupt(Object => :immediate){}
end
...
end
}
}
...
th.raise # stop thread Этот пример также можно записать следующим образом; такой вариант следует использовать, чтобы избежать асинхронных прерываний.
flag = true
th = Thread.new{
Thread.handle_interrupt(RuntimeError => :on_blocking){
while true
...
# reach safe point to invoke interrupt
break if flag == false
...
end
}
}
...
flag = false # stop thread static VALUE
rb_thread_s_report_exc(VALUE _)
{
return RBOOL(GET_THREAD()->vm->thread_report_on_exception);
} Возвращает состояние глобального параметра «сообщать об исключении».
Начиная с Ruby 2.5 значение по умолчанию — true.
Все потоки, созданные при включённом флаге, выведут сообщение в $stderr, если исключение завершит поток.
Thread.new { 1.times { raise } }
в $stderr будет выведено следующее сообщение:
#<Thread:...> terminated with exception (report_on_exception is true):
Traceback (most recent call last):
2: from -e:1:in `block in <main>'
1: from -e:1:in `times' Это сделано для раннего обнаружения ошибок в потоках. В некоторых случаях такой вывод может быть нежелателен. Есть несколько способов избежать лишнего вывода:
-
Если исключение возникло непреднамеренно, лучше всего устранить его причину, чтобы оно больше не повторялось.
-
Если исключение ожидаемо, лучше перехватить его ближе к месту возникновения, а не позволять ему завершить
Thread. -
Если гарантируется, что для
Threadбудет вызванThread#joinилиThread#value, безопасно отключить этот вывод с помощьюThread.current.report_on_exception = falseпри запускеThread. Однако в этом случае исключение может быть обработано значительно позже или не обработано вовсе, если дляThreadтак и не будет вызван join, например из-за блокировки родительского потока.
См. также ::report_on_exception=.
Для задания этого параметра отдельному потоку существует также метод уровня экземпляра; см. report_on_exception=.
static VALUE
rb_thread_s_report_exc_set(VALUE self, VALUE val)
{
GET_THREAD()->vm->thread_report_on_exception = RTEST(val);
return val;
} Возвращает новое состояние. Если задано значение true, все потоки, созданные впоследствии, унаследуют этот параметр и выведут сообщение в $stderr, если исключение завершит поток:
Thread.report_on_exception = true t1 = Thread.new do puts "In new thread" raise "Exception from thread" end sleep(1) puts "In the main thread"
В результате будет выведено:
In new thread #<Thread:...prog.rb:2> terminated with exception (report_on_exception is true): Traceback (most recent call last): prog.rb:4:in `block in <main>': Exception from thread (RuntimeError) In the main thread
См. также ::report_on_exception.
Для задания этого параметра отдельному потоку существует также метод уровня экземпляра; см. report_on_exception=.
static VALUE
thread_start(VALUE klass, VALUE args)
{
struct thread_create_params params = {
.type = thread_invoke_type_proc,
.args = args,
.proc = rb_block_proc(),
};
return thread_create_core(rb_thread_alloc(klass), ¶ms);
} По сути, то же, что и ::new. Однако если класс Thread является суперклассом, вызов start в этом подклассе не вызовет метод initialize подкласса.
static VALUE
thread_stop(VALUE _)
{
return rb_thread_stop();
} Останавливает выполнение текущего потока, переводит его в состояние «ожидания» и планирует выполнение другого потока.
a = Thread.new { print "a"; Thread.stop; print "c" }
sleep 0.1 while a.status!='sleep'
print "b"
a.run
a.join
#=> "abc"
Общедоступные методы экземпляра
static VALUE
rb_thread_aref(VALUE thread, VALUE key)
{
ID id = rb_check_id(&key);
if (!id) return Qnil;
return rb_thread_local_aref(thread, id);
} Обращение к атрибуту — возвращает значение локальной переменной волокна (корневого волокна текущего потока, если выполнение явно не находится внутри Fiber), используя имя в виде символа или строки. Если указанная переменная не существует, возвращает nil.
[
Thread.new { Thread.current["name"] = "A" },
Thread.new { Thread.current[:name] = "B" },
Thread.new { Thread.current["name"] = "C" }
].each do |th|
th.join
puts "#{th.inspect}: #{th[:name]}"
end
Результат:
#<Thread:0x00000002a54220 dead>: A #<Thread:0x00000002a541a8 dead>: B #<Thread:0x00000002a54130 dead>: C
Thread#[] и Thread#[]= являются локальными для волокна, а не для потока. В Ruby 1.8 такой путаницы не было, поскольку волокна появились только в Ruby 1.9. В Ruby 1.9 эти методы стали локальными для волокна, чтобы сохранить приведённый ниже идиоматический способ задания динамической области видимости.
def meth(newvalue)
begin
oldvalue = Thread.current[:name]
Thread.current[:name] = newvalue
yield
ensure
Thread.current[:name] = oldvalue
end
end
Этот приём может не работать как динамическая область видимости, если методы являются локальными для потока, а данный блок переключает волокно.
f = Fiber.new {
meth(1) {
Fiber.yield
}
}
meth(2) {
f.resume
}
f.resume
p Thread.current[:name]
#=> nil if fiber-local
#=> 2 if thread-local (The value 2 is leaked to outside of meth method.)
Сведения о локальных переменных потока см. в описаниях thread_variable_get и thread_variable_set.
static VALUE
rb_thread_aset(VALUE self, VALUE id, VALUE val)
{
return rb_thread_local_aset(self, rb_to_id(id), val);
} Присваивание атрибуту — задаёт или создаёт значение локальной переменной волокна, используя имя в виде символа или строки.
См. также Thread#[].
Сведения о локальных переменных потока см. в описаниях thread_variable_set и thread_variable_get.
static VALUE
rb_thread_abort_exc(VALUE thread)
{
return RBOOL(rb_thread_ptr(thread)->abort_on_exception);
} Возвращает состояние условия «прерывать при исключении», заданного для этого thr.
По умолчанию — false.
См. также abort_on_exception=.
Существует также метод уровня класса, задающий это значение для всех потоков; см. ::abort_on_exception.
static VALUE
rb_thread_abort_exc_set(VALUE thread, VALUE val)
{
rb_thread_ptr(thread)->abort_on_exception = RTEST(val);
return val;
} Если задано значение true и этот thr прерывается из-за исключения, возникшее исключение будет повторно вызвано в главном потоке.
См. также abort_on_exception.
Существует также метод уровня класса, задающий это значение для всех потоков; см. ::abort_on_exception=.
static VALUE
thread_add_trace_func_m(VALUE obj, VALUE trace)
{
thread_add_trace_func(GET_EC(), rb_thread_ptr(obj), trace);
return trace;
} Добавляет proc в качестве обработчика трассировки.
static VALUE
rb_thread_alive_p(VALUE thread)
{
return RBOOL(!thread_finished(rb_thread_ptr(thread)));
} Возвращает true, если thr выполняется или находится в спящем состоянии.
thr = Thread.new { }
thr.join #=> #<Thread:0x401b3fb0 dead>
Thread.current.alive? #=> true
thr.alive? #=> false
static VALUE
rb_thread_backtrace_m(int argc, VALUE *argv, VALUE thval)
{
return rb_vm_thread_backtrace(argc, argv, thval);
} Возвращает текущую трассировку стека целевого потока.
static VALUE
rb_thread_backtrace_locations_m(int argc, VALUE *argv, VALUE thval)
{
return rb_vm_thread_backtrace_locations(argc, argv, thval);
} Возвращает стек выполнения целевого потока — массив объектов, содержащих местоположения в трассировке стека.
Дополнительные сведения см. в Thread::Backtrace::Location.
Этот метод работает аналогично Kernel#caller_locations, но применяется к определённому потоку.
Завершает thr и передаёт выполнение другому потоку, возвращая завершённый Thread. Если это главный поток или последний поток, процесс завершается.
static VALUE
rb_thread_fetch(int argc, VALUE *argv, VALUE self)
{
VALUE key, val;
ID id;
rb_thread_t *target_th = rb_thread_ptr(self);
int block_given;
rb_check_arity(argc, 1, 2);
key = argv[0];
block_given = rb_block_given_p();
if (block_given && argc == 2) {
rb_warn("block supersedes default value argument");
}
id = rb_check_id(&key);
if (id == recursive_key) {
return target_th->ec->local_storage_recursive_hash;
}
else if (id && target_th->ec->local_storage &&
rb_id_table_lookup(target_th->ec->local_storage, id, &val)) {
return val;
}
else if (block_given) {
return rb_yield(key);
}
else if (argc == 1) {
rb_key_err_raise(rb_sprintf("key not found: %+"PRIsVALUE, key), self, key);
}
else {
return argv[1];
}
} Возвращает значение локальной переменной волокна для указанного ключа. Если ключ не найден, возможны несколько вариантов: если другие аргументы не указаны, будет вызвано исключение KeyError; если задан default, будет возвращено его значение; если указан необязательный блок кода, он будет выполнен, а его результат возвращён. См. Thread#[] и Hash#fetch.
VALUE
rb_thread_group(VALUE thread)
{
return rb_thread_ptr(thread)->thgroup;
} Возвращает ThreadGroup, содержащую данный поток.
Thread.main.group #=> #<ThreadGroup:0x4029d914>
static VALUE
thread_join_m(int argc, VALUE *argv, VALUE self)
{
VALUE timeout = Qnil;
rb_hrtime_t rel = 0, *limit = 0;
if (rb_check_arity(argc, 0, 1)) {
timeout = argv[0];
}
// Convert the timeout eagerly, so it's always converted and deterministic
/*
* This supports INFINITY and negative values, so we can't use
* rb_time_interval right now...
*/
if (NIL_P(timeout)) {
/* unlimited */
}
else if (FIXNUM_P(timeout)) {
rel = rb_sec2hrtime(NUM2TIMET(timeout));
limit = &rel;
}
else {
limit = double2hrtime(&rel, rb_num2dbl(timeout));
}
return thread_join(rb_thread_ptr(self), timeout, limit);
} Вызывающий поток приостанавливает выполнение и передаёт его этому thr.
Метод не возвращает управление, пока thr не завершится или пока не истечёт указанное время в limit секунд.
Если время ожидания истекло, возвращается nil, в противном случае возвращается thr.
Все потоки, к которым не было выполнено присоединение, будут уничтожены при завершении главной программы.
Если thr ранее вызвал исключение, а флаги ::abort_on_exception или $DEBUG не установлены (поэтому исключение ещё не было обработано), оно будет обработано в этот момент.
a = Thread.new { print "a"; sleep(10); print "b"; print "c" }
x = Thread.new { print "x"; Thread.pass; print "y"; print "z" }
x.join # Let thread x finish, thread a will be killed on exit.
#=> "axyz"
В следующем примере показано использование параметра limit.
y = Thread.new { 4.times { sleep 0.1; puts 'tick... ' }}
puts "Waiting" until y.join(0.15)
Результат:
tick... Waiting tick... Waiting tick... tick...
static VALUE
rb_thread_key_p(VALUE self, VALUE key)
{
VALUE val;
ID id = rb_check_id(&key);
struct rb_id_table *local_storage = rb_thread_ptr(self)->ec->local_storage;
if (!id || local_storage == NULL) {
return Qfalse;
}
return RBOOL(rb_id_table_lookup(local_storage, id, &val));
} Возвращает true, если указанная строка (или символ) существует в качестве локальной переменной волокна.
me = Thread.current me[:oliver] = "a" me.key?(:oliver) #=> true me.key?(:stanley) #=> false
static VALUE
rb_thread_keys(VALUE self)
{
struct rb_id_table *local_storage = rb_thread_ptr(self)->ec->local_storage;
VALUE ary = rb_ary_new();
if (local_storage) {
rb_id_table_foreach(local_storage, thread_keys_i, (void *)ary);
}
return ary;
} Возвращает массив имён локальных переменных волокна (в виде символов).
thr = Thread.new do Thread.current[:cat] = 'meow' Thread.current["dog"] = 'woof' end thr.join #=> #<Thread:0x401b3f10 dead> thr.keys #=> [:dog, :cat]
VALUE
rb_thread_kill(VALUE thread)
{
rb_thread_t *target_th = rb_thread_ptr(thread);
if (target_th->to_kill || target_th->status == THREAD_KILLED) {
return thread;
}
if (target_th == target_th->vm->ractor.main_thread) {
rb_exit(EXIT_SUCCESS);
}
RUBY_DEBUG_LOG("target_th:%u", rb_th_serial(target_th));
if (target_th == GET_THREAD()) {
/* kill myself immediately */
rb_threadptr_to_kill(target_th);
}
else {
threadptr_check_pending_interrupt_queue(target_th);
rb_threadptr_pending_interrupt_enque(target_th, RUBY_FATAL_THREAD_KILLED);
rb_threadptr_interrupt(target_th);
}
return thread;
} Завершает thr и передаёт выполнение другому потоку, возвращая завершённый Thread. Если это главный поток или последний поток, процесс завершается.
static VALUE
rb_thread_getname(VALUE thread)
{
return rb_thread_ptr(thread)->name;
} Показывает имя потока.
static VALUE
rb_thread_setname(VALUE thread, VALUE name)
{
rb_thread_t *target_th = rb_thread_ptr(thread);
if (!NIL_P(name)) {
rb_encoding *enc;
StringValueCStr(name);
enc = rb_enc_get(name);
if (!rb_enc_asciicompat(enc)) {
rb_raise(rb_eArgError, "ASCII incompatible encoding (%s)",
rb_enc_name(enc));
}
name = rb_str_new_frozen(name);
}
target_th->name = name;
if (threadptr_initialized(target_th) && target_th->has_dedicated_nt) {
native_set_another_thread_name(target_th->nt->thread_id, name);
}
return name;
} Задаёт указанное имя потоку Ruby. На некоторых платформах имя может быть задано для pthread и/или ядра операционной системы.
static VALUE
rb_thread_native_thread_id(VALUE thread)
{
rb_thread_t *target_th = rb_thread_ptr(thread);
if (rb_threadptr_dead(target_th)) return Qnil;
return native_thread_native_thread_id(target_th);
} Возвращает идентификатор системного потока, используемый потоком Ruby.
Идентификатор зависит от операционной системы. (Это не идентификатор потока POSIX, возвращаемый pthread_self(3).)
-
В Linux это TID, возвращаемый gettid(2).
-
В macOS это уникальный для всей системы целочисленный идентификатор потока, возвращаемый pthread_threadid_np(3).
-
Во FreeBSD это уникальный целочисленный идентификатор потока, возвращаемый pthread_getthreadid_np(3).
-
В Windows это идентификатор потока, возвращаемый GetThreadId().
-
На других платформах вызывается исключение
NotImplementedError.
ПРИМЕЧАНИЕ: если поток ещё не связан с системным потоком или уже отвязан от него, возвращается nil. Если в реализации Ruby используется модель потоков M:N, идентификатор может меняться в зависимости от момента вызова.
static VALUE
rb_thread_pending_interrupt_p(int argc, VALUE *argv, VALUE target_thread)
{
rb_thread_t *target_th = rb_thread_ptr(target_thread);
if (!target_th->pending_interrupt_queue) {
return Qfalse;
}
if (rb_threadptr_pending_interrupt_empty_p(target_th)) {
return Qfalse;
}
if (rb_check_arity(argc, 0, 1)) {
VALUE err = argv[0];
if (!rb_obj_is_kind_of(err, rb_cModule)) {
rb_raise(rb_eTypeError, "class or module required for rescue clause");
}
return RBOOL(rb_threadptr_pending_interrupt_include_p(target_th, err));
}
else {
return Qtrue;
}
} Возвращает признак того, пуста ли очередь асинхронных событий целевого потока.
Если задано error, проверяются только отложенные события типа error.
Дополнительные сведения см. в ::pending_interrupt?.
static VALUE
rb_thread_priority(VALUE thread)
{
return INT2NUM(rb_thread_ptr(thread)->priority);
} Возвращает приоритет thr. По умолчанию он наследуется от текущего потока, создавшего новый поток, либо равен нулю для исходного главного потока; потоки с более высоким приоритетом выполняются чаще, чем потоки с более низким приоритетом (однако потоки с более низким приоритетом также могут выполняться).
Это лишь подсказка планировщику потоков Ruby. На некоторых платформах она может игнорироваться.
Thread.current.priority #=> 0
static VALUE
rb_thread_priority_set(VALUE thread, VALUE prio)
{
rb_thread_t *target_th = rb_thread_ptr(thread);
int priority;
#if USE_NATIVE_THREAD_PRIORITY
target_th->priority = NUM2INT(prio);
native_thread_apply_priority(th);
#else
priority = NUM2INT(prio);
if (priority > RUBY_THREAD_PRIORITY_MAX) {
priority = RUBY_THREAD_PRIORITY_MAX;
}
else if (priority < RUBY_THREAD_PRIORITY_MIN) {
priority = RUBY_THREAD_PRIORITY_MIN;
}
target_th->priority = (int8_t)priority;
#endif
return INT2NUM(target_th->priority);
} Устанавливает приоритет thr равным integer. Потоки с более высоким приоритетом выполняются чаще, чем потоки с более низким приоритетом (однако потоки с более низким приоритетом также могут выполняться).
Это лишь подсказка планировщику потоков Ruby. На некоторых платформах она может игнорироваться.
count1 = count2 = 0
a = Thread.new do
loop { count1 += 1 }
end
a.priority = -1
b = Thread.new do
loop { count2 += 1 }
end
b.priority = -2
sleep 1 #=> 1
count1 #=> 622504
count2 #=> 5832
static VALUE
thread_raise_m(int argc, VALUE *argv, VALUE self)
{
rb_thread_t *target_th = rb_thread_ptr(self);
const rb_thread_t *current_th = GET_THREAD();
threadptr_check_pending_interrupt_queue(target_th);
rb_threadptr_raise(target_th, argc, argv);
/* To perform Thread.current.raise as Kernel.raise */
if (current_th == target_th) {
RUBY_VM_CHECK_INTS(target_th->ec);
}
return Qnil;
} Вызывает исключение из указанного потока. Вызывающий поток не обязан быть thr. Дополнительные сведения см. в Kernel#raise.
Thread.abort_on_exception = true
a = Thread.new { sleep(200) }
a.raise("Gotcha")
Результат:
prog.rb:3: Gotcha (RuntimeError) from prog.rb:2:in `initialize' from prog.rb:2:in `new' from prog.rb:2
static VALUE
rb_thread_report_exc(VALUE thread)
{
return RBOOL(rb_thread_ptr(thread)->report_on_exception);
} Возвращает состояние условия «сообщать об исключении», заданного для этого thr.
Значение по умолчанию при создании Thread совпадает со значением глобального флага Thread.report_on_exception.
См. также report_on_exception=.
Существует также метод уровня класса, задающий это значение для всех новых потоков; см. ::report_on_exception=.
static VALUE
rb_thread_report_exc_set(VALUE thread, VALUE val)
{
rb_thread_ptr(thread)->report_on_exception = RTEST(val);
return val;
} Если задано значение true, при завершении этого thr из-за исключения в $stderr выводится сообщение. Подробности см. в ::report_on_exception.
См. также report_on_exception.
Существует также метод уровня класса, задающий это значение для всех новых потоков; см. ::report_on_exception=.
VALUE
rb_thread_run(VALUE thread)
{
rb_thread_wakeup(thread);
rb_thread_schedule();
return thread;
} Пробуждает thr, делая его доступным для планирования.
a = Thread.new { puts "a"; Thread.stop; puts "c" }
sleep 0.1 while a.status!='sleep'
puts "Got here"
a.run
a.join
Результат:
a Got here c
См. также метод экземпляра wakeup.
static VALUE
thread_set_trace_func_m(VALUE target_thread, VALUE trace)
{
rb_execution_context_t *ec = GET_EC();
rb_thread_t *target_th = rb_thread_ptr(target_thread);
rb_threadptr_remove_event_hook(ec, target_th, call_trace_func, Qundef);
if (NIL_P(trace)) {
return Qnil;
}
else {
thread_add_trace_func(ec, target_th, trace);
return trace;
}
} Назначает proc обработчиком трассировки для thr или отключает трассировку, если параметр равен nil.
static VALUE
rb_thread_status(VALUE thread)
{
rb_thread_t *target_th = rb_thread_ptr(thread);
if (rb_threadptr_dead(target_th)) {
if (!NIL_P(target_th->ec->errinfo) &&
!FIXNUM_P(target_th->ec->errinfo)) {
return Qnil;
}
else {
return Qfalse;
}
}
else {
return rb_str_new2(thread_status_name(target_th, FALSE));
}
} Возвращает состояние thr.
"sleep"-
Возвращается, если поток находится в спящем состоянии или ожидает ввода-вывода
"run"-
Поток выполняется
"aborting"-
Поток прерывается
false-
Поток завершён обычным образом
nil-
Поток завершён из-за исключения.
a = Thread.new { raise("die now") }
b = Thread.new { Thread.stop }
c = Thread.new { Thread.exit }
d = Thread.new { sleep }
d.kill #=> #<Thread:0x401b3678 aborting>
a.status #=> nil
b.status #=> "sleep"
c.status #=> false
d.status #=> "aborting"
Thread.current.status #=> "run"
static VALUE
rb_thread_stop_p(VALUE thread)
{
rb_thread_t *th = rb_thread_ptr(thread);
if (rb_threadptr_dead(th)) {
return Qtrue;
}
return RBOOL(th->status == THREAD_STOPPED || th->status == THREAD_STOPPED_FOREVER);
} Возвращает true, если thr завершён или находится в спящем состоянии.
a = Thread.new { Thread.stop }
b = Thread.current
a.stop? #=> true
b.stop? #=> false
Завершает thr и передаёт выполнение другому потоку, возвращая завершённый Thread. Если это главный поток или последний поток, процесс завершается.
static VALUE
rb_thread_variable_p(VALUE thread, VALUE key)
{
VALUE locals;
VALUE symbol = rb_to_symbol(key);
if (LIKELY(!THREAD_LOCAL_STORAGE_INITIALISED_P(thread))) {
return Qfalse;
}
locals = rb_thread_local_storage(thread);
return RBOOL(rb_hash_lookup(locals, symbol) != Qnil);
} Возвращает true, если указанная строка (или символ) существует в качестве локальной переменной потока.
me = Thread.current me.thread_variable_set(:oliver, "a") me.thread_variable?(:oliver) #=> true me.thread_variable?(:stanley) #=> false
Обратите внимание, что это не локальные переменные волокна. Дополнительные сведения см. в описаниях Thread#[] и Thread#thread_variable_get.
static VALUE
rb_thread_variable_get(VALUE thread, VALUE key)
{
VALUE locals;
VALUE symbol = rb_to_symbol(key);
if (LIKELY(!THREAD_LOCAL_STORAGE_INITIALISED_P(thread))) {
return Qnil;
}
locals = rb_thread_local_storage(thread);
return rb_hash_aref(locals, symbol);
} Возвращает значение заданной локальной переменной потока. Обратите внимание, что эти переменные отличаются от локальных значений волокна. Сведения о локальных значениях волокна см. в описаниях Thread#[] и Thread#[]=.
Локальные значения Thread сохраняются вместе с потоками и не зависят от волокон. Например:
Thread.new {
Thread.current.thread_variable_set("foo", "bar") # set a thread local
Thread.current["foo"] = "bar" # set a fiber local
Fiber.new {
Fiber.yield [
Thread.current.thread_variable_get("foo"), # get the thread local
Thread.current["foo"], # get the fiber local
]
}.resume
}.join.value # => ['bar', nil]
Для локальной переменной потока возвращается значение «bar», а для локальной переменной волокна — nil. Волокно выполняется в том же потоке, поэтому локальные значения потока доступны.
static VALUE
rb_thread_variable_set(VALUE thread, VALUE key, VALUE val)
{
VALUE locals;
if (OBJ_FROZEN(thread)) {
rb_frozen_error_raise(thread, "can't modify frozen thread locals");
}
locals = rb_thread_local_storage(thread);
return rb_hash_aset(locals, rb_to_symbol(key), val);
} Задаёт локальную переменную потока с именем key и значением value. Обратите внимание, что эти переменные локальны для потоков, а не для волокон. Дополнительные сведения см. в описаниях Thread#thread_variable_get и Thread#[].
static VALUE
rb_thread_variables(VALUE thread)
{
VALUE locals;
VALUE ary;
ary = rb_ary_new();
if (LIKELY(!THREAD_LOCAL_STORAGE_INITIALISED_P(thread))) {
return ary;
}
locals = rb_thread_local_storage(thread);
rb_hash_foreach(locals, keys_i, ary);
return ary;
} Возвращает массив имён локальных переменных потока (в виде символов).
thr = Thread.new do
Thread.current.thread_variable_set(:cat, 'meow')
Thread.current.thread_variable_set("dog", 'woof')
end
thr.join #=> #<Thread:0x401b3f10 dead>
thr.thread_variables #=> [:dog, :cat]
Обратите внимание, что это не локальные переменные волокна. Дополнительные сведения см. в описаниях Thread#[] и Thread#thread_variable_get.
static VALUE
rb_thread_to_s(VALUE thread)
{
VALUE cname = rb_class_path(rb_obj_class(thread));
rb_thread_t *target_th = rb_thread_ptr(thread);
const char *status;
VALUE str, loc;
status = thread_status_name(target_th, TRUE);
str = rb_sprintf("#<%"PRIsVALUE":%p", cname, (void *)thread);
if (!NIL_P(target_th->name)) {
rb_str_catf(str, "@%"PRIsVALUE, target_th->name);
}
if ((loc = threadptr_invoke_proc_location(target_th)) != Qnil) {
rb_str_catf(str, " %"PRIsVALUE":%"PRIsVALUE,
RARRAY_AREF(loc, 0), RARRAY_AREF(loc, 1));
}
rb_str_catf(str, " %s>", status);
return str;
} Выводит имя, идентификатор и состояние thr в виде строки.
static VALUE
thread_value(VALUE self)
{
rb_thread_t *th = rb_thread_ptr(self);
thread_join(th, Qnil, 0);
if (UNDEF_P(th->value)) {
// If the thread is dead because we forked th->value is still Qundef.
return Qnil;
}
return th->value;
} Ожидает завершения thr с помощью join, затем возвращает его значение или вызывает исключение, из-за которого завершился поток.
a = Thread.new { 2 + 2 }
a.value #=> 4
b = Thread.new { raise 'something went wrong' }
b.value #=> RuntimeError: something went wrong
VALUE
rb_thread_wakeup(VALUE thread)
{
if (!RTEST(rb_thread_wakeup_alive(thread))) {
rb_raise(rb_eThreadError, "killed thread");
}
return thread;
} Помечает указанный поток как доступный для планирования, однако он всё ещё может оставаться заблокированным на операциях ввода-вывода.
Примечание: это не вызывает планировщик; дополнительные сведения см. в run.
c = Thread.new { Thread.stop; puts "hey!" }
sleep 0.1 while c.status!='sleep'
c.wakeup
c.join
#=> "hey!"
Ruby Core © 1993–2025 Yukihiro Matsumoto
Licensed under the Ruby License.
Ruby Standard Library © contributors
Licensed under their own licenses.