diff --git a/.github/CONTRIBUTING.md b/.github/CONTRIBUTING.md index a865b48b3..0f6450e5c 100644 --- a/.github/CONTRIBUTING.md +++ b/.github/CONTRIBUTING.md @@ -66,7 +66,7 @@ There are a few guidelines which we follow when adding features. Consider that s #### Write Documentation -Document any external behavior in the [README](README.md). +Document any external behavior in the [README](../README.md). #### Commit Changes @@ -106,7 +106,7 @@ git push origin my-feature-branch -f #### Update CHANGELOG -Update the [CHANGELOG](CHANGELOG.md) with a description of what you have changed. +Update the [CHANGELOG](../CHANGELOG.md) with a description of what you have changed. #### Check on Your Pull Request diff --git a/.github/release.yml b/.github/release.yml new file mode 100644 index 000000000..db1d8e962 --- /dev/null +++ b/.github/release.yml @@ -0,0 +1,4 @@ +changelog: + exclude: + labels: + - dependencies # Added by Dependabot diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index ab1d23db8..26a2679b4 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -6,16 +6,17 @@ concurrency: jobs: build: + name: "Tests: Ruby ${{ matrix.ruby }} - ${{ matrix.os }}" runs-on: ${{ matrix.os }}-latest timeout-minutes: 10 strategy: fail-fast: false matrix: - ruby: [2.3, 2.4, 2.5, 2.6, 2.7, '3.0', 3.1, 3.2, 3.3, jruby, truffleruby] + ruby: [2.3, 2.4, 2.5, 2.6, 2.7, '3.0', 3.1, 3.2, 3.3, 3.4, jruby, truffleruby] os: [ubuntu] include: - - ruby: 3.3 + - ruby: ruby os: windows env: @@ -23,10 +24,9 @@ jobs: RUBYOPT: '-w' JRUBY_OPTS: '--dev' - name: "Tests: Ruby ${{ matrix.ruby }} - ${{ matrix.os }}" steps: - name: Clone Repo - uses: actions/checkout@v4 + uses: actions/checkout@v6 - name: Setup Ruby ${{ matrix.ruby }} uses: ruby/setup-ruby@v1 with: @@ -35,6 +35,21 @@ jobs: - name: Run tests run: bundle exec rake ci + no-extensions: + name: "Test without C extension" + runs-on: ubuntu-latest + timeout-minutes: 10 + env: + RUBYOPT: '-w' + steps: + - uses: actions/checkout@v6 + - uses: ruby/setup-ruby@v1 + with: + ruby-version: ruby + bundler-cache: true + - name: Run tests + run: bundle exec rake spec:ci + isolated: name: "Test isolated" runs-on: ubuntu-latest @@ -46,7 +61,7 @@ jobs: env: RUBYOPT: '-w' steps: - - uses: actions/checkout@v4 + - uses: actions/checkout@v6 - uses: ruby/setup-ruby@v1 with: ruby-version: ${{ matrix.ruby }} diff --git a/.github/workflows/docs.yml b/.github/workflows/docs.yml index 780f0741c..6f3954070 100644 --- a/.github/workflows/docs.yml +++ b/.github/workflows/docs.yml @@ -18,7 +18,7 @@ jobs: env: BUNDLE_WITH: "documentation" steps: - - uses: actions/checkout@v4 + - uses: actions/checkout@v6 with: fetch-depth: 0 - uses: ruby/setup-ruby@v1 @@ -29,7 +29,7 @@ jobs: - run: ruby support/generate_docs.rb - name: Upload artifact - uses: actions/upload-pages-artifact@v3 + uses: actions/upload-pages-artifact@v4 with: path: docs diff --git a/.github/workflows/experimental.yml b/.github/workflows/experimental.yml index 28048747c..f3aa626af 100644 --- a/.github/workflows/experimental.yml +++ b/.github/workflows/experimental.yml @@ -10,6 +10,7 @@ jobs: timeout-minutes: 10 strategy: + fail-fast: false matrix: ruby: [head, jruby-head, truffleruby-head] @@ -21,7 +22,7 @@ jobs: name: "Tests: Experimental Ruby ${{ matrix.ruby }}" steps: - name: Clone Repo - uses: actions/checkout@v4 + uses: actions/checkout@v6 - name: Setup Ruby ${{ matrix.ruby }} uses: ruby/setup-ruby@v1 with: diff --git a/CHANGELOG.md b/CHANGELOG.md index 30810b724..549b9c186 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,11 @@ ## Current +## Release v1.3.6 (13 December 2025) + +concurrent-ruby: + +* See the [release notes on GitHub](https://github.com/ruby-concurrency/concurrent-ruby/releases/tag/v1.3.6). + ## Release v1.3.5, edge v0.7.2 (15 January 2025) concurrent-ruby: diff --git a/README.md b/README.md index 63ee743c5..daf3eac5a 100644 --- a/README.md +++ b/README.md @@ -207,7 +207,7 @@ Deprecated features are still available and bugs are being fixed, but new featur These are available in the `concurrent-ruby-edge` companion gem. These features are under active development and may change frequently. They are expected not to -keep backward compatibility (there may also lack tests and documentation). Semantic versions will +keep backward compatibility (they may also lack tests and documentation). Semantic versions will be obeyed though. Features developed in `concurrent-ruby-edge` are expected to move to `concurrent-ruby` when final. @@ -358,7 +358,8 @@ best practice is to depend on `concurrent-ruby` and let users to decide if they * Recent CRuby * JRuby, `rbenv install jruby-9.2.17.0` * Set env variable `CONCURRENT_JRUBY_HOME` to point to it, e.g. `/usr/local/opt/rbenv/versions/jruby-9.2.17.0` -* Install Docker, required for Windows builds +* Install Docker or Podman, required for Windows builds +* If `bundle config get path` is set, use `bundle config set --local path.system true` otherwise the `gem name, path: '.'` gems won't be found (Bundler limitation). ### Publishing the Gem @@ -378,6 +379,7 @@ best practice is to depend on `concurrent-ruby` and let users to decide if they * [Charles Oliver Nutter](https://github.com/headius) * [Ben Sheldon](https://github.com/bensheldon) * [Samuel Williams](https://github.com/ioquatix) +* [Joshua Young](https://github.com/joshuay03) ### Special Thanks to diff --git a/docs-source/signpost.md b/docs-source/signpost.md index 7b52f16ac..83b009e34 100644 --- a/docs-source/signpost.md +++ b/docs-source/signpost.md @@ -3,7 +3,7 @@ Pick a `concurrent-ruby` version: * [master](./master/index.html) -* [1.3.5 with edge 0.7.2](./1.3.5/index.html) +* [1.3.6 with edge 0.7.2](./1.3.6/index.html) * [1.1.10 with edge 0.6.0](./1.1.10/index.html) * [1.1.9 with edge 0.6.0](./1.1.9/index.html) * [1.1.8 with edge 0.6.0](./1.1.8/index.html) diff --git a/ext/concurrent-ruby-ext/atomic_boolean.c b/ext/concurrent-ruby-ext/atomic_boolean.c index d4913df56..cdd684245 100644 --- a/ext/concurrent-ruby-ext/atomic_boolean.c +++ b/ext/concurrent-ruby-ext/atomic_boolean.c @@ -7,8 +7,16 @@ void atomic_boolean_mark(void *value) { rb_gc_mark_maybe((VALUE) value); } +const rb_data_type_t atomic_boolean_type = { + "Concurrent::CAtomicBoolean", + { + atomic_boolean_mark, + RUBY_NEVER_FREE, + }, +}; + VALUE atomic_boolean_allocate(VALUE klass) { - return rb_data_object_wrap(klass, (void *) Qfalse, atomic_boolean_mark, NULL); + return rb_data_typed_object_wrap(klass, (void *) Qfalse, &atomic_boolean_type); } VALUE method_atomic_boolean_initialize(int argc, VALUE* argv, VALUE self) { diff --git a/ext/concurrent-ruby-ext/atomic_fixnum.c b/ext/concurrent-ruby-ext/atomic_fixnum.c index faa0ac0ac..088874006 100644 --- a/ext/concurrent-ruby-ext/atomic_fixnum.c +++ b/ext/concurrent-ruby-ext/atomic_fixnum.c @@ -7,8 +7,16 @@ void atomic_fixnum_mark(void *value) { rb_gc_mark_maybe((VALUE) value); } +const rb_data_type_t atomic_fixnum_type = { + "Concurrent::CAtomicFixnum", + { + atomic_fixnum_mark, + RUBY_NEVER_FREE, + }, +}; + VALUE atomic_fixnum_allocate(VALUE klass) { - return rb_data_object_wrap(klass, (void *) Qnil, atomic_fixnum_mark, NULL); + return rb_data_typed_object_wrap(klass, (void *) Qnil, &atomic_fixnum_type); } VALUE method_atomic_fixnum_initialize(int argc, VALUE* argv, VALUE self) { diff --git a/ext/concurrent-ruby-ext/atomic_reference.c b/ext/concurrent-ruby-ext/atomic_reference.c index d99e5afd5..abdbebb7e 100644 --- a/ext/concurrent-ruby-ext/atomic_reference.c +++ b/ext/concurrent-ruby-ext/atomic_reference.c @@ -43,8 +43,16 @@ void ir_mark(void *value) { rb_gc_mark_maybe((VALUE) value); } +const rb_data_type_t ir_type = { + "Concurrent::CAtomicReference", + { + ir_mark, + RUBY_NEVER_FREE, + }, +}; + VALUE ir_alloc(VALUE klass) { - return rb_data_object_wrap(klass, (void *) Qnil, ir_mark, NULL); + return rb_data_typed_object_wrap(klass, (void *) Qnil, &ir_type); } VALUE ir_initialize(int argc, VALUE* argv, VALUE self) { @@ -78,7 +86,11 @@ VALUE ir_get_and_set(VALUE self, VALUE new_value) { } VALUE ir_compare_and_set(volatile VALUE self, VALUE expect_value, VALUE new_value) { -#if defined(__ENVIRONMENT_MAC_OS_X_VERSION_MIN_REQUIRED__) && __ENVIRONMENT_MAC_OS_X_VERSION_MIN_REQUIRED__ >= 1050 +#if defined(__ENVIRONMENT_MAC_OS_X_VERSION_MIN_REQUIRED__) && __ENVIRONMENT_MAC_OS_X_VERSION_MIN_REQUIRED__ >= 101200 + if (atomic_compare_exchange_strong_explicit((_Atomic uintptr_t *)&DATA_PTR(self), &expect_value, new_value, memory_order_seq_cst, memory_order_seq_cst)) { + return Qtrue; + } +#elif defined(__ENVIRONMENT_MAC_OS_X_VERSION_MIN_REQUIRED__) && __ENVIRONMENT_MAC_OS_X_VERSION_MIN_REQUIRED__ >= 1050 if (OSAtomicCompareAndSwap64(expect_value, new_value, &DATA_PTR(self))) { return Qtrue; } diff --git a/ext/concurrent-ruby-ext/atomic_reference.h b/ext/concurrent-ruby-ext/atomic_reference.h index 1f33414a9..34136759c 100644 --- a/ext/concurrent-ruby-ext/atomic_reference.h +++ b/ext/concurrent-ruby-ext/atomic_reference.h @@ -5,6 +5,9 @@ #include #endif +#if defined(__ENVIRONMENT_MAC_OS_X_VERSION_MIN_REQUIRED__) && __ENVIRONMENT_MAC_OS_X_VERSION_MIN_REQUIRED__ >= 101200 +#include +#endif #ifdef HAVE_LIBKERN_OSATOMIC_H #include #endif diff --git a/lib/concurrent-ruby/concurrent/atomic/lock_local_var.rb b/lib/concurrent-ruby/concurrent/atomic/lock_local_var.rb index ebf23a241..c367d2622 100644 --- a/lib/concurrent-ruby/concurrent/atomic/lock_local_var.rb +++ b/lib/concurrent-ruby/concurrent/atomic/lock_local_var.rb @@ -6,6 +6,7 @@ module Concurrent # @!visibility private def self.mutex_owned_per_thread? return false if Concurrent.on_jruby? || Concurrent.on_truffleruby? + return RUBY_VERSION < "3.0" if Concurrent.on_cruby? mutex = Mutex.new # Lock the mutex: diff --git a/lib/concurrent-ruby/concurrent/collection/ruby_timeout_queue.rb b/lib/concurrent-ruby/concurrent/collection/ruby_timeout_queue.rb new file mode 100644 index 000000000..3cf53718f --- /dev/null +++ b/lib/concurrent-ruby/concurrent/collection/ruby_timeout_queue.rb @@ -0,0 +1,55 @@ +module Concurrent + module Collection + # @!visibility private + # @!macro ruby_timeout_queue + class RubyTimeoutQueue < ::Queue + def initialize(*args) + if RUBY_VERSION >= '3.2' + raise "#{self.class.name} is not needed on Ruby 3.2 or later, use ::Queue instead" + end + + super(*args) + + @mutex = Mutex.new + @cond_var = ConditionVariable.new + end + + def push(obj) + @mutex.synchronize do + super(obj) + @cond_var.signal + end + end + alias_method :enq, :push + alias_method :<<, :push + + def pop(non_block = false, timeout: nil) + if non_block && timeout + raise ArgumentError, "can't set a timeout if non_block is enabled" + end + + if non_block + super(true) + elsif timeout + @mutex.synchronize do + deadline = Concurrent.monotonic_time + timeout + while (now = Concurrent.monotonic_time) < deadline && empty? + @cond_var.wait(@mutex, deadline - now) + end + begin + return super(true) + rescue ThreadError + # still empty + nil + end + end + else + super(false) + end + end + alias_method :deq, :pop + alias_method :shift, :pop + end + private_constant :RubyTimeoutQueue + end +end diff --git a/lib/concurrent-ruby/concurrent/collection/timeout_queue.rb b/lib/concurrent-ruby/concurrent/collection/timeout_queue.rb new file mode 100644 index 000000000..7048c4b68 --- /dev/null +++ b/lib/concurrent-ruby/concurrent/collection/timeout_queue.rb @@ -0,0 +1,18 @@ +module Concurrent + module Collection + # @!visibility private + # @!macro internal_implementation_note + TimeoutQueueImplementation = if RUBY_VERSION >= '3.2' + ::Queue + else + require 'concurrent/collection/ruby_timeout_queue' + RubyTimeoutQueue + end + private_constant :TimeoutQueueImplementation + + # @!visibility private + # @!macro timeout_queue + class TimeoutQueue < TimeoutQueueImplementation + end + end +end diff --git a/lib/concurrent-ruby/concurrent/executor/fixed_thread_pool.rb b/lib/concurrent-ruby/concurrent/executor/fixed_thread_pool.rb index 8324c0673..b73cff877 100644 --- a/lib/concurrent-ruby/concurrent/executor/fixed_thread_pool.rb +++ b/lib/concurrent-ruby/concurrent/executor/fixed_thread_pool.rb @@ -81,10 +81,8 @@ module Concurrent # What is being pruned is controlled by the min_threads and idletime # parameters passed at pool creation time # - # This is a no-op on some pool implementation (e.g. the Java one). The Ruby - # pool will auto-prune each time a new job is posted. You will need to call - # this method explicitly in case your application post jobs in bursts (a - # lot of jobs and then nothing for long periods) + # This is a no-op on all pool implementations as they prune themselves + # automatically, and has been deprecated. # @!macro thread_pool_executor_public_api # diff --git a/lib/concurrent-ruby/concurrent/executor/java_executor_service.rb b/lib/concurrent-ruby/concurrent/executor/java_executor_service.rb index b2bc69a6e..96991a78e 100644 --- a/lib/concurrent-ruby/concurrent/executor/java_executor_service.rb +++ b/lib/concurrent-ruby/concurrent/executor/java_executor_service.rb @@ -46,6 +46,7 @@ def shutdown def kill synchronize do @executor.shutdownNow + wait_for_termination nil end end diff --git a/lib/concurrent-ruby/concurrent/executor/java_thread_pool_executor.rb b/lib/concurrent-ruby/concurrent/executor/java_thread_pool_executor.rb index 598a5f91f..a89fde27d 100644 --- a/lib/concurrent-ruby/concurrent/executor/java_thread_pool_executor.rb +++ b/lib/concurrent-ruby/concurrent/executor/java_thread_pool_executor.rb @@ -8,6 +8,7 @@ module Concurrent # @!macro thread_pool_options # @!visibility private class JavaThreadPoolExecutor < JavaExecutorService + include Concern::Deprecation # @!macro thread_pool_executor_constant_default_max_pool_size DEFAULT_MAX_POOL_SIZE = java.lang.Integer::MAX_VALUE # 2147483647 @@ -100,6 +101,7 @@ def running? # @!macro thread_pool_executor_method_prune_pool def prune_pool + deprecated "#prune_pool has no effect and will be removed in the next release." end private diff --git a/lib/concurrent-ruby/concurrent/executor/ruby_single_thread_executor.rb b/lib/concurrent-ruby/concurrent/executor/ruby_single_thread_executor.rb index 916337d4b..6d9f88c2c 100644 --- a/lib/concurrent-ruby/concurrent/executor/ruby_single_thread_executor.rb +++ b/lib/concurrent-ruby/concurrent/executor/ruby_single_thread_executor.rb @@ -1,4 +1,5 @@ require 'concurrent/executor/ruby_thread_pool_executor' +require 'concurrent/executor/serial_executor_service' module Concurrent @@ -6,6 +7,7 @@ module Concurrent # @!macro abstract_executor_service_public_api # @!visibility private class RubySingleThreadExecutor < RubyThreadPoolExecutor + include SerialExecutorService # @!macro single_thread_executor_method_initialize def initialize(opts = {}) diff --git a/lib/concurrent-ruby/concurrent/executor/ruby_thread_pool_executor.rb b/lib/concurrent-ruby/concurrent/executor/ruby_thread_pool_executor.rb index 9375acf38..28bbdcad7 100644 --- a/lib/concurrent-ruby/concurrent/executor/ruby_thread_pool_executor.rb +++ b/lib/concurrent-ruby/concurrent/executor/ruby_thread_pool_executor.rb @@ -3,6 +3,7 @@ require 'concurrent/concern/logging' require 'concurrent/executor/ruby_executor_service' require 'concurrent/utility/monotonic_time' +require 'concurrent/collection/timeout_queue' module Concurrent @@ -10,6 +11,7 @@ module Concurrent # @!macro thread_pool_options # @!visibility private class RubyThreadPoolExecutor < RubyExecutorService + include Concern::Deprecation # @!macro thread_pool_executor_constant_default_max_pool_size DEFAULT_MAX_POOL_SIZE = 2_147_483_647 # java.lang.Integer::MAX_VALUE @@ -94,9 +96,28 @@ def remaining_capacity end end + # removes the worker if it can be pruned + # + # @return [true, false] if the worker was pruned + # # @!visibility private - def remove_busy_worker(worker) - synchronize { ns_remove_busy_worker worker } + def prune_worker(worker) + synchronize do + if ns_prunable_capacity > 0 + remove_worker worker + true + else + false + end + end + end + + # @!visibility private + def remove_worker(worker) + synchronize do + ns_remove_ready_worker worker + ns_remove_busy_worker worker + end end # @!visibility private @@ -116,7 +137,7 @@ def worker_task_completed # @!macro thread_pool_executor_method_prune_pool def prune_pool - synchronize { ns_prune_pool } + deprecated "#prune_pool has no effect and will be removed in next the release, see https://github.com/ruby-concurrency/concurrent-ruby/pull/1082." end private @@ -146,9 +167,6 @@ def ns_initialize(opts) @largest_length = 0 @workers_counter = 0 @ruby_pid = $$ # detects if Ruby has forked - - @gc_interval = opts.fetch(:gc_interval, @idletime / 2.0).to_i # undocumented - @next_gc_time = Concurrent.monotonic_time + @gc_interval end # @!visibility private @@ -162,12 +180,10 @@ def ns_execute(*args, &task) if ns_assign_worker(*args, &task) || ns_enqueue(*args, &task) @scheduled_task_count += 1 + nil else - return fallback_action(*args, &task) + fallback_action(*args, &task) end - - ns_prune_pool if @next_gc_time < Concurrent.monotonic_time - nil end # @!visibility private @@ -218,7 +234,7 @@ def ns_assign_worker(*args, &task) # @!visibility private def ns_enqueue(*args, &task) return false if @synchronous - + if !ns_limited_queue? || @queue.size < @max_queue @queue << [task, args] true @@ -265,7 +281,7 @@ def ns_ready_worker(worker, last_message, success = true) end end - # removes a worker which is not in not tracked in @ready + # removes a worker which is not tracked in @ready # # @!visibility private def ns_remove_busy_worker(worker) @@ -274,25 +290,27 @@ def ns_remove_busy_worker(worker) true end - # try oldest worker if it is idle for enough time, it's returned back at the start - # # @!visibility private - def ns_prune_pool - now = Concurrent.monotonic_time - stopped_workers = 0 - while !@ready.empty? && (@pool.size - stopped_workers > @min_length) - worker, last_message = @ready.first - if now - last_message > self.idletime - stopped_workers += 1 - @ready.shift - worker << :stop - else break - end + def ns_remove_ready_worker(worker) + if index = @ready.index { |rw, _| rw == worker } + @ready.delete_at(index) end + true + end - @next_gc_time = Concurrent.monotonic_time + @gc_interval + # @return [Integer] number of excess idle workers which can be removed without + # going below min_length, or all workers if not running + # + # @!visibility private + def ns_prunable_capacity + if running? + [@pool.size - @min_length, @ready.size].min + else + @pool.size + end end + # @!visibility private def ns_reset_if_forked if $$ != @ruby_pid @queue.clear @@ -312,7 +330,7 @@ class Worker def initialize(pool, id) # instance variables accessed only under pool's lock so no need to sync here again - @queue = Queue.new + @queue = Collection::TimeoutQueue.new @pool = pool @thread = create_worker @queue, pool, pool.idletime @@ -338,17 +356,22 @@ def kill def create_worker(queue, pool, idletime) Thread.new(queue, pool, idletime) do |my_queue, my_pool, my_idletime| catch(:stop) do - loop do + prunable = true - case message = my_queue.pop + loop do + timeout = prunable && my_pool.running? ? my_idletime : nil + case message = my_queue.pop(timeout: timeout) + when nil + throw :stop if my_pool.prune_worker(self) + prunable = false when :stop - my_pool.remove_busy_worker(self) + my_pool.remove_worker(self) throw :stop - else task, args = message run_task my_pool, task, args my_pool.ready_worker(self, Concurrent.monotonic_time) + prunable = true end end end diff --git a/lib/concurrent-ruby/concurrent/executor/timer_set.rb b/lib/concurrent-ruby/concurrent/executor/timer_set.rb index 759dce093..b936c7b2f 100644 --- a/lib/concurrent-ruby/concurrent/executor/timer_set.rb +++ b/lib/concurrent-ruby/concurrent/executor/timer_set.rb @@ -61,6 +61,7 @@ def post(delay, *args, &task) # not running. def kill shutdown + @timer_executor.kill end private :<< @@ -122,7 +123,9 @@ def remove_task(task) def ns_shutdown_execution ns_reset_if_forked @queue.clear - @timer_executor.kill + @condition.set + @condition.reset + @timer_executor.shutdown stopped_event.set end diff --git a/lib/concurrent-ruby/concurrent/executors.rb b/lib/concurrent-ruby/concurrent/executors.rb index eb1972ce6..5d5cd391b 100644 --- a/lib/concurrent-ruby/concurrent/executors.rb +++ b/lib/concurrent-ruby/concurrent/executors.rb @@ -10,7 +10,6 @@ require 'concurrent/executor/ruby_executor_service' require 'concurrent/executor/ruby_single_thread_executor' require 'concurrent/executor/ruby_thread_pool_executor' -require 'concurrent/executor/cached_thread_pool' require 'concurrent/executor/safe_task_executor' require 'concurrent/executor/serial_executor_service' require 'concurrent/executor/serialized_execution' diff --git a/lib/concurrent-ruby/concurrent/mvar.rb b/lib/concurrent-ruby/concurrent/mvar.rb index dfc41950c..9777ba347 100644 --- a/lib/concurrent-ruby/concurrent/mvar.rb +++ b/lib/concurrent-ruby/concurrent/mvar.rb @@ -9,7 +9,7 @@ module Concurrent # queue of length one, or a special kind of mutable variable. # # On top of the fundamental `#put` and `#take` operations, we also provide a - # `#mutate` that is atomic with respect to operations on the same instance. + # `#modify` that is atomic with respect to operations on the same instance. # These operations all support timeouts. # # We also support non-blocking operations `#try_put!` and `#try_take!`, a @@ -87,7 +87,7 @@ def borrow(timeout = nil) @mutex.synchronize do wait_for_full(timeout) - # if we timeoud out we'll still be empty + # If we timed out we'll still be empty if unlocked_full? yield @value else @@ -116,10 +116,10 @@ def put(value, timeout = nil) end # Atomically `take`, yield the value to a block for transformation, and then - # `put` the transformed value. Returns the transformed value. A timeout can + # `put` the transformed value. Returns the pre-transform value. A timeout can # be set to limit the time spent blocked, in which case it returns `TIMEOUT` # if the time is exceeded. - # @return [Object] the transformed value, or `TIMEOUT` + # @return [Object] the pre-transform value, or `TIMEOUT` def modify(timeout = nil) raise ArgumentError.new('no block given') unless block_given? diff --git a/lib/concurrent-ruby/concurrent/promise.rb b/lib/concurrent-ruby/concurrent/promise.rb index c717f9b0d..8df80b0c7 100644 --- a/lib/concurrent-ruby/concurrent/promise.rb +++ b/lib/concurrent-ruby/concurrent/promise.rb @@ -167,7 +167,7 @@ module Concurrent # c2 = p.then(-> reason { raise 'Boom!' }) # # c1.wait.state #=> :fulfilled - # c1.value #=> 45 + # c1.value #=> 42 # c2.wait.state #=> :rejected # c2.reason #=> # # ``` diff --git a/lib/concurrent-ruby/concurrent/timer_task.rb b/lib/concurrent-ruby/concurrent/timer_task.rb index dd2037f62..d1eb55699 100644 --- a/lib/concurrent-ruby/concurrent/timer_task.rb +++ b/lib/concurrent-ruby/concurrent/timer_task.rb @@ -2,6 +2,7 @@ require 'concurrent/concern/dereferenceable' require 'concurrent/concern/observable' require 'concurrent/atomic/atomic_boolean' +require 'concurrent/atomic/atomic_fixnum' require 'concurrent/executor/executor_service' require 'concurrent/executor/ruby_executor_service' require 'concurrent/executor/safe_task_executor' @@ -236,6 +237,7 @@ def execute synchronize do if @running.false? @running.make_true + @age.increment schedule_next_task(@run_now ? 0 : @execution_interval) end end @@ -309,6 +311,7 @@ def ns_initialize(opts, &task) @task = Concurrent::SafeTaskExecutor.new(task) @executor = opts[:executor] || Concurrent.global_io_executor @running = Concurrent::AtomicBoolean.new(false) + @age = Concurrent::AtomicFixnum.new(0) @value = nil self.observers = Collection::CopyOnNotifyObserverSet.new @@ -328,13 +331,15 @@ def ns_kill_execution # @!visibility private def schedule_next_task(interval = execution_interval) - ScheduledTask.execute(interval, executor: @executor, args: [Concurrent::Event.new], &method(:execute_task)) + ScheduledTask.execute(interval, executor: @executor, args: [Concurrent::Event.new, @age.value], &method(:execute_task)) nil end # @!visibility private - def execute_task(completion) + def execute_task(completion, age_when_scheduled) return nil unless @running.true? + return nil unless @age.value == age_when_scheduled + start_time = Concurrent.monotonic_time _success, value, reason = @task.execute(self) if completion.try? diff --git a/lib/concurrent-ruby/concurrent/version.rb b/lib/concurrent-ruby/concurrent/version.rb index f773e44f1..6833ed4d9 100644 --- a/lib/concurrent-ruby/concurrent/version.rb +++ b/lib/concurrent-ruby/concurrent/version.rb @@ -1,3 +1,3 @@ module Concurrent - VERSION = '1.3.5' + VERSION = '1.3.6' end diff --git a/spec/concurrent/atomic/atomic_boolean_spec.rb b/spec/concurrent/atomic/atomic_boolean_spec.rb index 133b7f2ce..76edd3e27 100644 --- a/spec/concurrent/atomic/atomic_boolean_spec.rb +++ b/spec/concurrent/atomic/atomic_boolean_spec.rb @@ -142,7 +142,7 @@ module Concurrent end end - if Concurrent.allow_c_extensions? + if Concurrent.c_extensions_loaded? RSpec.describe CAtomicBoolean do it_should_behave_like :atomic_boolean end @@ -165,7 +165,7 @@ module Concurrent it 'inherits from JavaAtomicBoolean' do expect(AtomicBoolean.ancestors).to include(JavaAtomicBoolean) end - elsif Concurrent.allow_c_extensions? + elsif Concurrent.c_extensions_loaded? it 'inherits from CAtomicBoolean' do expect(AtomicBoolean.ancestors).to include(CAtomicBoolean) end diff --git a/spec/concurrent/atomic/atomic_fixnum_spec.rb b/spec/concurrent/atomic/atomic_fixnum_spec.rb index 39259b13f..8222bf523 100644 --- a/spec/concurrent/atomic/atomic_fixnum_spec.rb +++ b/spec/concurrent/atomic/atomic_fixnum_spec.rb @@ -204,7 +204,7 @@ module Concurrent end end - if Concurrent.allow_c_extensions? + if Concurrent.c_extensions_loaded? RSpec.describe CAtomicFixnum do it_should_behave_like :atomic_fixnum end @@ -227,7 +227,7 @@ module Concurrent it 'inherits from JavaAtomicFixnum' do expect(AtomicFixnum.ancestors).to include(JavaAtomicFixnum) end - elsif Concurrent.allow_c_extensions? + elsif Concurrent.c_extensions_loaded? it 'inherits from CAtomicFixnum' do expect(AtomicFixnum.ancestors).to include(CAtomicFixnum) end diff --git a/spec/concurrent/atomic/atomic_reference_spec.rb b/spec/concurrent/atomic/atomic_reference_spec.rb index 2cc0caada..7eaf2e55e 100644 --- a/spec/concurrent/atomic/atomic_reference_spec.rb +++ b/spec/concurrent/atomic/atomic_reference_spec.rb @@ -163,7 +163,7 @@ module Concurrent it_should_behave_like :atomic_reference end - if Concurrent.allow_c_extensions? + if Concurrent.c_extensions_loaded? RSpec.describe CAtomicReference do it_should_behave_like :atomic_reference end @@ -190,7 +190,7 @@ module Concurrent it 'inherits from JavaAtomicReference' do expect(described_class.ancestors).to include(Concurrent::JavaAtomicReference) end - elsif Concurrent.allow_c_extensions? + elsif Concurrent.c_extensions_loaded? it 'inherits from CAtomicReference' do expect(described_class.ancestors).to include(Concurrent::CAtomicReference) end diff --git a/spec/concurrent/executor/cached_thread_pool_spec.rb b/spec/concurrent/executor/cached_thread_pool_spec.rb index 513873b92..e56b258c2 100644 --- a/spec/concurrent/executor/cached_thread_pool_spec.rb +++ b/spec/concurrent/executor/cached_thread_pool_spec.rb @@ -152,15 +152,13 @@ module Concurrent context 'garbage collection' do - subject { described_class.new(idletime: 0.1, max_threads: 2, gc_interval: 0) } + subject { described_class.new(idletime: 0.1, max_threads: 2) } it 'removes from pool any thread that has been idle too long' do latch = Concurrent::CountDownLatch.new(4) 4.times { subject.post { sleep 0.1; latch.count_down } } + sleep 0.4 expect(latch.wait(1)).to be true - sleep 0.2 - subject.post {} - sleep 0.2 expect(subject.length).to be < 4 end @@ -197,25 +195,27 @@ module Concurrent expect(subject.length).to be >= 5 3.times { subject << proc { sleep(1) } } sleep(0.1) - expect(subject.length).to be >= 5 + expect(subject.length).to be >= 3 end end end context 'stress' do configurations = [ - { min_threads: 2, - max_threads: ThreadPoolExecutor::DEFAULT_MAX_POOL_SIZE, - idletime: 0.1, # 1 minute - max_queue: 0, # unlimited + { + min_threads: 2, + max_threads: ThreadPoolExecutor::DEFAULT_MAX_POOL_SIZE, + idletime: 60, # 1 minute + max_queue: 0, # unlimited fallback_policy: :caller_runs, # shouldn't matter -- 0 max queue - gc_interval: 0.1 }, - { min_threads: 2, - max_threads: 4, - idletime: 0.1, # 1 minute - max_queue: 0, # unlimited + }, + { + min_threads: 2, + max_threads: 4, + idletime: 60, # 1 minute + max_queue: 0, # unlimited fallback_policy: :caller_runs, # shouldn't matter -- 0 max queue - gc_interval: 0.1 } + } ] configurations.each do |config| diff --git a/spec/concurrent/executor/java_thread_pool_executor_spec.rb b/spec/concurrent/executor/java_thread_pool_executor_spec.rb index 64d445441..4d447317b 100644 --- a/spec/concurrent/executor/java_thread_pool_executor_spec.rb +++ b/spec/concurrent/executor/java_thread_pool_executor_spec.rb @@ -26,13 +26,6 @@ module Concurrent it_should_behave_like :thread_pool_executor - context :prune do - it "is a no-op, pruning is handled by the JVM" do - executor = JavaThreadPoolExecutor.new - executor.prune_pool - end - end - context '#overload_policy' do specify ':abort maps to AbortPolicy' do diff --git a/spec/concurrent/executor/ruby_thread_pool_executor_spec.rb b/spec/concurrent/executor/ruby_thread_pool_executor_spec.rb index bf328a7c6..de6ddccf0 100644 --- a/spec/concurrent/executor/ruby_thread_pool_executor_spec.rb +++ b/spec/concurrent/executor/ruby_thread_pool_executor_spec.rb @@ -56,68 +56,49 @@ def wakeup_thread_group(group) end before(:each) do - @now = Concurrent.monotonic_time - allow(Concurrent).to receive(:monotonic_time) { @now } - @group1 = prepare_thread_group(5) @group2 = prepare_thread_group(5) end def eventually(mutex: nil, timeout: 5, &block) - start = Time.now - while Time.now - start < timeout - begin - if mutex - mutex.synchronize do - return yield - end - else + start = Time.now + while Time.now - start < timeout + begin + if mutex + mutex.synchronize do return yield end - rescue Exception => last_failure + else + return yield end - Thread.pass + rescue Exception => last_failure end - raise last_failure - end - - it "triggers pruning when posting work if the last prune happened more than gc_interval ago" do - wakeup_thread_group(@group1) - @now += 6 - wakeup_thread_group(@group2) - subject.post { } - - eventually { expect(@group1.threads).to all(have_attributes(status: false)) } - eventually { expect(@group2.threads).to all(have_attributes(status: 'sleep')) } + Thread.pass + end + raise last_failure end - it "does not trigger pruning when posting work if the last prune happened less than gc_interval ago" do + it "triggers pruning if the thread idletimes have elapsed" do wakeup_thread_group(@group1) - @now += 3 - subject.prune_pool - @now += 3 + sleep 6 wakeup_thread_group(@group2) - subject.post { } eventually { expect(@group1.threads).to all(have_attributes(status: false)) } eventually { expect(@group2.threads).to all(have_attributes(status: 'sleep')) } end - it "reclaims threads that have been idle for more than idletime seconds" do + it "does not trigger pruning if the thread idletimes have not elapsed " do wakeup_thread_group(@group1) - @now += 6 wakeup_thread_group(@group2) - subject.prune_pool - eventually { expect(@group1.threads).to all(have_attributes(status: false)) } + eventually { expect(@group1.threads).to all(have_attributes(status: 'sleep')) } eventually { expect(@group2.threads).to all(have_attributes(status: 'sleep')) } end it "keeps at least min_length workers" do wakeup_thread_group(@group1) wakeup_thread_group(@group2) - @now += 12 - subject.prune_pool + sleep 6 all_threads = @group1.threads + @group2.threads eventually do finished_threads = all_threads.find_all { |t| !t.status } diff --git a/spec/concurrent/executor/thread_pool_shared.rb b/spec/concurrent/executor/thread_pool_shared.rb index a8ab3e246..9e5f72c12 100644 --- a/spec/concurrent/executor/thread_pool_shared.rb +++ b/spec/concurrent/executor/thread_pool_shared.rb @@ -86,16 +86,15 @@ it 'returns zero on creation' do expect(subject.completed_task_count).to eq 0 end - - unless Concurrent.on_jruby? - - it 'returns the approximate number of tasks that have been completed thus far' do - 5.times{ subject.post{ raise StandardError } } - 5.times{ subject.post{ nil } } - subject.post { latch.count_down } - latch.wait(1) - expect(subject.completed_task_count).to be > 0 - end + + it 'returns the approximate number of tasks that have been completed thus far' do + 5.times{ subject.post{ raise StandardError } } + 5.times{ subject.post{ nil } } + subject.post { latch.count_down } + latch.wait(1) + subject.shutdown + expect(subject.wait_for_termination(pool_termination_timeout)).to eq true + expect(subject.completed_task_count).to be > 0 end end diff --git a/spec/concurrent/executor/timer_set_spec.rb b/spec/concurrent/executor/timer_set_spec.rb index a386553b4..f4169473d 100644 --- a/spec/concurrent/executor/timer_set_spec.rb +++ b/spec/concurrent/executor/timer_set_spec.rb @@ -244,7 +244,7 @@ module Concurrent expect(task.schedule_time).to eq expected end - it 'reschdules a pending and unpost task when given a valid time' do + it 'reschedules a pending and unpost task when given a valid time' do initial_delay = 10 rescheduled_delay = 20 task = subject.post(initial_delay){ nil } diff --git a/spec/concurrent/timer_task_spec.rb b/spec/concurrent/timer_task_spec.rb index 10fbf34db..dbb5d43d6 100644 --- a/spec/concurrent/timer_task_spec.rb +++ b/spec/concurrent/timer_task_spec.rb @@ -1,5 +1,6 @@ require_relative 'concern/dereferenceable_shared' require_relative 'concern/observable_shared' +require 'concurrent/atomic/atomic_fixnum' require 'concurrent/timer_task' module Concurrent @@ -116,6 +117,19 @@ def trigger_observable(observable) sleep(0.1) expect(task.shutdown).to be_truthy end + + it 'will cancel pre-shutdown task even if restarted to avoid double-runs' do + counter = Concurrent::AtomicFixnum.new(0) + task = TimerTask.execute(execution_interval: 0.2, run_now: true) { counter.increment } + sleep 0.05 + expect(counter.value).to eq 1 + + task.shutdown + task.execute + + sleep 0.25 + expect(counter.value).to eq 3 + end end end