An Efficient Algorithm for Concurrent Priority Queue Heaps

Pseudocode from article of the above name. by Galen C. Hunt, Maged M. Michael, Srinivasan Parthasarthy, and Michael L. Scott.  Incorporates a bug fix due to Anton Malakhov of Intel.

Data Structures and Macros:

  structure data_item_t begin
      lock := FREE
      tag := EMPTY
      priority := 0
  end

  structure heap_t begin
      lock := FREE
      bit_reversed_counter_t size
      data_item_t items[]
  end

  define LOCK(x) as lock(heap.items[x].lock)
  define UNLOCK(x) as unlock(heap.items[x].lock)
  define TAG(x) as heap.items[x].tag
  define PRIORITY(x) as heap.items[x].priority
  

Insert Operation

  procedure concurrent_insert(priority, heap: heap_t)
      // Insert new item at bottom of the heap.
      lock(heap.lock)
      i := bit_reversed_increment(heap.size)
      LOCK(i)
      unlock(heap.lock)
      PRIORITY(i) := priority
      TAG(i) := pid
      UNLOCK(i)

      // Move item towards top of heap while it has a higher priority
      //   than its parent.
      while i > 1 do
          parent := i / 2
          LOCK(parent)
          LOCK(i)
          old_i := i
          if TAG(parent) = AVAILABLE and TAG(i) = pid then
              if PRIORITY(i) > PRIORITY(parent) then
                  swap_items(i, parent)
                  i := parent
              else
                  TAG(i) := AVAILABLE
                  i := 0
              endif
          else if TAG(parent) = EMPTY then
              i := 0
          else if TAG(i) != pid then
              i := parent
          endif

          UNLOCK(old_i)
          UNLOCK(parent)
      enddo
      if i = 1 then
          LOCK(i)
          if TAG(i) = pid then
              TAG(i) := AVAILABLE
          endif
          UNLOCK(i)
      endif
  end
  

Delete Operation

  function concurrent_delete(heap: heap_t)
      // Grab an item from the bottom of the heap to replace the
      //   to-be-deleted top item.
      lock(heap.lock)
      bottom := bit_reversed_decrement(heap.size)
      LOCK(bottom)
      unlock(heap.lock)
      priority := PRIORITY(bottom)
      TAG(bottom) := EMPTY
      UNLOCK(bottom)

      // Lock first item.  Stop if it was the only item in the heap.
      LOCK(1)
      if TAG(1) = EMPTY then
          UNLOCK(1)
          return priority
      endif

      // Replace the top item with the item stored from the bottom.
      swap(priority, PRIORITY(1))
      TAG(1) := AVAILABLE

      // Adjust the heap starting at the top.
      //   We always hold a lock on the item being adjusted.
      i := 1
      while (i < MAX_SIZE / 2) do
          left := i * 2
          right := i * 2 + 1
          LOCK(left)
          LOCK(right)
          if TAG(left) = EMPTY then
              UNLOCK(right)
              UNLOCK(left)
              break
          else if TAG(right) = EMPTY or PRIORITY(left) > PRIORITY(right) then
              UNLOCK(right)
              child := left
          else
              UNLOCK(left)
              child := right
          endif

          // If the child has a higher priority than the parent then
          //   swap them.  If not, stop.
          if PRIORITY(child) > PRIORITY(i) then
              swap_items(child, i)
              UNLOCK(i)
              i := child
          else
              UNLOCK(child)
              break
          endif
      enddo
      UNLOCK(i)
      return priority
  end
  

Bit-Reversed Counter

  structure bit_reversed_counter begin
      counter := 0
      reversed := 0
      high_bit := -1
  end

  function bit_reversed_increment(c: bit_reversed_counter_t)
      c.counter := c.counter + 1

      for bit := c.high_bit - 1 to 0 step -1
          c.reversed := not(c.reversed, bit)
          if test(c.reversed, bit) = TRUE then
              break
          endif
      endfor

      if bit < 0 then
          c.reversed := c.counter
          c.high_bit := c.high_bit + 1
      endif

      return c.reversed
  end

  function bit_reversed_decrement(c: bit_reversed_counter_t)
      c.counter := c.counter - 1

      for bit := c.high_bit - 1 to 0 step -1
          c.reversed := not(c.reversed, bit)
          if test(c.reversed, bit) = FALSE then
              break
          endif
      endfor

      if bit < 0 then
          c.reversed := c.counter
          c.high_bit := c.high_bit - 1
      endif

      return c.reversed
  end