Method try_read_lock has a Cognitive Complexity of 7 (exceeds 5 allowed). Consider refactoring.

Open

    def try_read_lock
      if (held = @HeldCount.value) > 0
        if held & READ_LOCK_MASK == 0
          # If we hold a write lock, but not a read lock...
          @Counter.update { |c| c + 1 }

• Read up
  Read up

Method sink has a Cognitive Complexity of 7 (exceeds 5 allowed). Consider refactoring.

Open

      def sink(k)
        success = false

        while (j = (2 * k)) <= @length do
          j += 1 if j < @length && ! ordered?(j, j+1)

• Read up
  Read up

Method try_acquire has a Cognitive Complexity of 7 (exceeds 5 allowed). Consider refactoring.

Open

    def try_acquire
      while true
        current_capacity = capacity
        if current_capacity > 0
          return true if compare_and_set_capacity(

• Read up
  Read up

Method fetch_or_store has a Cognitive Complexity of 7 (exceeds 5 allowed). Consider refactoring.

Open

    def fetch_or_store(key, default_value = NULL)
      fetch(key) do
        put(key, block_given? ? yield(key) : (NULL == default_value ? raise_fetch_no_key : default_value))
      end
    end

• Read up
  Read up

Method execute_task_once has a Cognitive Complexity of 7 (exceeds 5 allowed). Consider refactoring.

Open

    def execute_task_once # :nodoc:
      # this function has been optimized for performance and
      # should not be modified without running new benchmarks
      execute = task = nil
      synchronize do

• Read up
  Read up

Method aggregate has a Cognitive Complexity of 7 (exceeds 5 allowed). Consider refactoring.

Open

    def self.aggregate(method, *promises)
      composite = Promise.new do
        completed = promises.collect do |promise|
          promise.execute if promise.unscheduled?
          promise.wait

• Read up
  Read up

Method post has a Cognitive Complexity of 7 (exceeds 5 allowed). Consider refactoring.

Open

    def post(*args, &task)
      raise ArgumentError.new('no block given') unless block_given?
      return false unless running?
      @count.increment
      Thread.new(*args) do

• Read up
  Read up

Method add_observer has a Cognitive Complexity of 7 (exceeds 5 allowed). Consider refactoring.

Open

    def add_observer(observer = nil, func = :update, &block)
      raise ArgumentError.new('cannot provide both an observer and a block') if observer && block
      direct_notification = false

      if block

• Read up
  Read up

Method await_for has a Cognitive Complexity of 7 (exceeds 5 allowed). Consider refactoring.

Open

      def await_for(timeout, *agents)
        end_at = Concurrent.monotonic_time + timeout.to_f
        ok     = agents.length.times do |i|
          break false if (delay = end_at - Concurrent.monotonic_time) < 0
          break false unless agents[i].await_for(delay)

• Read up
  Read up

Method call_dataflow has a Cognitive Complexity of 7 (exceeds 5 allowed). Consider refactoring.

Open

  def call_dataflow(method, executor, *inputs, &block)
    raise ArgumentError.new('an executor must be provided') if executor.nil?
    raise ArgumentError.new('no block given') unless block_given?
    unless inputs.all? { |input| input.is_a? IVar }
      raise ArgumentError.new("Not all dependencies are IVars.\nDependencies: #{ inputs.inspect }")

• Read up
  Read up

Method wait_while has a Cognitive Complexity of 7 (exceeds 5 allowed). Consider refactoring.

Open

    def wait_while(condition, timeout)
      if timeout.nil?
        while yield
          condition.wait(@mutex)
        end

• Read up
  Read up

Method on_envelope has a Cognitive Complexity of 7 (exceeds 5 allowed). Consider refactoring.

Open

      def on_envelope(envelope)
        log(DEBUG) { "is  #{envelope.future ? 'asked' : 'told'} #{envelope.message.inspect} by #{envelope.sender}" }
        schedule_execution do
          log(DEBUG) { "was #{envelope.future ? 'asked' : 'told'} #{envelope.message.inspect} by #{envelope.sender} - processing" }
          process_envelope envelope

• Read up
  Read up

Method []= has a Cognitive Complexity of 7 (exceeds 5 allowed). Consider refactoring.

Open

    def []=(member, value)
      if member.is_a? Integer
        length = synchronize { @values.length }
        if member >= length
          raise IndexError.new("offset #{member} too large for struct(size:#{length})")

• Read up
  Read up

Method ns_assign_worker has a Cognitive Complexity of 7 (exceeds 5 allowed). Consider refactoring.

Open

    def ns_assign_worker(*args, &task)
      # keep growing if the pool is not at the minimum yet
      worker, _ = (@ready.pop if @pool.size >= @min_length) || ns_add_busy_worker
      if worker
        worker << [task, args]

• Read up
  Read up

Method acquire_or_event has a Cognitive Complexity of 7 (exceeds 5 allowed). Consider refactoring.

Open

    def acquire_or_event
      while true
        current_capacity = capacity
        if compare_and_set_capacity current_capacity, current_capacity - 1
          if current_capacity > 0

• Read up
  Read up

Method ns_initialize has a Cognitive Complexity of 7 (exceeds 5 allowed). Consider refactoring.

Open

    def ns_initialize(initial, opts)
      @error_mode    = opts[:error_mode]
      @error_handler = opts[:error_handler]

      if @error_mode && !ERROR_MODES.include?(@error_mode)

• Read up
  Read up

Method terminate! has a Cognitive Complexity of 7 (exceeds 5 allowed). Consider refactoring.

Open

        def terminate!(reason = nil, envelope = nil)
          return true if terminated?

          self_termination = Concurrent::Promises.resolved_future(reason.nil?, reason.nil? || nil, reason)
          all_terminations = if @terminate_children

• Read up
  Read up

Method ns_get has a Cognitive Complexity of 7 (exceeds 5 allowed). Consider refactoring.

Open

      def ns_get(member)
        if member.is_a? Integer
          if member >= @values.length
            raise IndexError.new("offset #{member} too large for struct(size:#{@values.length})")
          end

• Read up
  Read up

Method put has a Cognitive Complexity of 7 (exceeds 5 allowed). Consider refactoring.

Open

        def put(item)
          mine = synchronize do
            return false if ns_closed?

            ref = Concurrent::AtomicReference.new(item)

• Read up
  Read up

Avoid too many return statements within this method.

Open

        return val;