# Purely functional octree and octree priority queue

I'm programming a game in Scala, and a lot of tasks are able to be done in the background. I've created a system in which each Future can be assigned a position in 3d space, and they are stored in an octree, which prioritizes them based on proximity to the player. I'd appreciate feedback on style and performance.

A bit of background, V3F is my class for a vector of 3 floats, and V3I is my subclass for a vector of 3 ints.

First, the purely functional octree:

import java.io.PrintStream

import com.phoenixkahlo.hellcraft.math._

import scala.collection.mutable.ArrayBuffer

object Signs {
val signs = Seq(V3I(1, 1, 1), V3I(-1, 1, 1), V3I(1, -1, 1), V3I(1, 1, -1), V3I(1, -1, -1), V3I(-1, 1, -1),
V3I(-1, -1, 1), V3I(-1, -1, -1))
def apply(): Seq[V3I] = signs
}

sealed trait Octree[+E] extends Map[V3F, E] {
override def +[V1 >: E](kv: (V3F, V1)): Octree[V1]

override def -(key: V3F): Octree[E]

def closest(point: V3F): Option[(V3F, E)]

def prettyPrint(indentation: Int = 0, out: PrintStream = System.out): Unit

def depth: Int
}

/**
* An octree with no contents
*/
case class EmptyOctree(center: V3F, range: Float) extends Octree[Nothing] {
override def +[V1 >: Nothing](kv: (V3F, V1)): Octree[V1] = OctreeLeaf(center, range, kv)

override def get(key: V3F): Option[Nothing] = None

override def iterator: Iterator[(V3F, Nothing)] = Iterator.empty

override def -(key: V3F): Octree[Nothing] = this

override def size: Int = 0

override def closest(point: V3F): Option[(V3F, Nothing)] = None

override def prettyPrint(indentation: Int, out: PrintStream): Unit = {
out.println("()")
}

override val depth = 0
}

/**
* An octree with one content
*/
case class OctreeLeaf[+E](center: V3F, range: Float, elem: (V3F, E)) extends Octree[E] {
override def +[V1 >: E](kv: (V3F, V1)): Octree[V1] = {
if (kv._1 == elem._1) OctreeLeaf(center, range, kv)
else {
val children = Signs().map(sign => sign -> EmptyOctree(center + (sign * range / 2), range / 2)).toMap
OctreeBranch(center, range, children) + elem + kv
}
}

override def get(key: V3F): Option[E] =
elem match {
case (v, e) if v == key => Some(e)
case _ => None
}

override def iterator: Iterator[(V3F, E)] = Iterator(elem)

override def -(key: V3F): Octree[E] =
elem match {
case (v, _) if v == key => EmptyOctree(center, range)
case _ => this
}

override def size: Int = 1

override def closest(point: V3F): Option[(V3F, E)] = Some(elem)

override def prettyPrint(indentation: Int, out: PrintStream): Unit = {
out.println(elem)
}

override val depth = 1
}

case class OctreeBranch[+E](center: V3F, range: Float, children: Map[V3I, Octree[E]]) extends Octree[E] {
override def +[V1 >: E](kv: (V3F, V1)): Octree[V1] = {
val (key, value) = kv
val sign = (key - center).map(n => if (n >= 0) 1 else -1).toInts
OctreeBranch(center, range, children.updated(sign, children(sign) + kv))
}

override def get(key: V3F): Option[E] = {
val sign = (key - center).map(n => if (n >= 0) 1 else -1).toInts
children(sign).get(key)
}

override def iterator: Iterator[(V3F, E)] =
children.values.flatten.iterator

override def -(key: V3F): Octree[E] = {
val sign = (key - center).map(n => if (n >= 0) 1 else -1).toInts
OctreeBranch(center, range, children.updated(sign, children(sign) - key))
}

override def size: Int = children.values.map(_.size).sum

override def closest(point: V3F): Option[(V3F, E)] = {
Signs().flatMap(children(_).closest(point)).sortBy(_._1 dist point).headOption
}

override def prettyPrint(indentation: Int, out: PrintStream): Unit = {
def justify(v: V3I): String = {
var s = v.toString
while (s.size < 12)
s += ' '
s
}

println('{')
for ((sign, child) <- children) {
for (_ <- 1 to (indentation + 2)) print(' ')
print(justify(sign) + " -> ")
child.prettyPrint(indentation + 2, out)
}
for (_ <- 1 to indentation) print(' '); println('}')
}

override lazy val depth = children.values.map(_.depth).max + 1
}


A basic priority queue using the octree:

/**
* Uses an octree to create a priority queue based of vector/generic pairs based on their proximity to an adjustable
* point. This is not thread safe, although internal variables are declared volatile.
*/
class OctreePriorityQueue[E] extends util.AbstractQueue[(V3F, E)] {

@volatile var point: V3F = Origin
@volatile var tree: Octree[E] = EmptyOctree(Origin, Float.MaxValue)

override def poll(): (V3F, E) = {
tree.closest(point) match {
case Some((k, v)) =>
tree -= k
(k, v)
case None => null
}
}

override def offer(e: (V3F, E)): Boolean = {
tree += e
true
}

override def peek(): (V3F, E) = {
tree.closest(point) match {
case Some(item) => item
case None => null
}
}

override def iterator(): util.Iterator[(V3F, E)] =
JavaConverters.asJavaIterator(tree.iterator)

override def size(): Int =
tree.size

}


An improved version that stores the items in bins so that multiple items can have the same position and go through in a FIFO manor:

class OctreeBinPriorityQueue[E] extends util.AbstractQueue[(V3F, E)] {

var point: V3F = Origin
var tree: Octree[Queue[E]] = EmptyOctree(Origin, Float.MaxValue)

override def poll(): (V3F, E) = {
tree.closest(point) match {
case Some((key, queue)) =>
val (value, newQueue) = queue.dequeue
if (newQueue isEmpty) tree -= key
else tree += key -> newQueue
_size -= 1
(key, value)
case None => null
}
}

override def offer(e: (V3F, E)): Boolean = {
val (key, value) = e
val queue = tree.getOrElse(key, Queue.empty)
val newQueue = queue.enqueue(value)
tree += key -> newQueue
_size += 1
true
}

override def peek(): (V3F, E) = {
tree.closest(point) match {
case Some((k, q)) => k -> q.head
case None => null
}
}

override def iterator(): util.Iterator[(V3F, E)] =
JavaConverters.asJavaIterator(tree.iterator.flatMap({ case (k, q) => q.map(k -> _) }))

@volatile private var _size: Int = 0

override def size(): Int = _size

}


Finally, a layer over the OctreeBinPriorityQueue that makes it into a BlockingQueue:

/**
* Implements a concurrency layer on top of an OctreePriorityQueue to make it a BlockingQueue.
*
* This is implemented using a ReentrantReadWriteLock, for reading and writing to the internal queue, and a
* LinkedBlockingQueue of "tickets", which only holds references to a "ticket" singleton. Whenever a thread adds an
* item to the queue, it then adds a ticket to the ticket queue. Whenever a thread wishes to remove an item from
* the queue, it first takes a ticket from the ticket queue. The reading and writing of the wrapped queue
* is still protected by the lock, but the ticket taking and giving is not. The purpose of this system is to
* delegate the blocking logic a blocking system which is already established to work efficiently.
*/
class OctreeBlockingQueue[E] extends util.AbstractQueue[(V3F, E)] with util.concurrent.BlockingQueue[(V3F, E)] {
private val queue = new OctreeBinPriorityQueue[E]
private val lock = new ReentrantReadWriteLock
private val writeLock = lock.writeLock()
private val readLock = lock.readLock()
private object Ticket
private val tickets = new LinkedBlockingQueue[Object]

def point = {
try {
readLock.lock()
queue.point
} finally readLock.unlock()
}

def point_=(p: V3F) = {
try {
writeLock.lock()
queue.point = p
} finally writeLock.unlock()
}

override def poll(): (V3F, E) = {
if (tickets.poll() == null) null
else try {
writeLock.lock()
queue.remove()
} finally writeLock.unlock()
}

override def poll(timeout: Long, unit: TimeUnit): (V3F, E) = this.poll()

override def add(e: (V3F, E)): Boolean = {
try {
writeLock.lock()
queue.add(e)
} finally writeLock.unlock()
tickets.add(Ticket)
}

override def put(e: (V3F, E)): Unit = this.add(e)

override def offer(e: (V3F, E)): Boolean = this.add(e)

override def offer(e: (V3F, E), timeout: Long, unit: TimeUnit): Boolean = this.add(e)

override def drainTo(c: util.Collection[_ >: (V3F, E)], maxElements: Int): Int = {
try {
readLock.lock()
var item: (V3F, E) = null
var count = 0
while (count < maxElements && {item = poll(); item} != null) {
c.add(item)
count += 1
}
count
} finally readLock.unlock()
}

override def drainTo(c: util.Collection[_ >: (V3F, E)]): Int = this.drainTo(c, Int.MaxValue)

override def take(): (V3F, E) = {
tickets.take()
try {
writeLock.lock()
queue.remove()
} finally writeLock.unlock()
}

override def remainingCapacity(): Int = Int.MaxValue

override def peek(): (V3F, E) = {
try {
readLock.lock()
queue.peek()
} finally readLock.unlock()
}

/**
* Since the OctreeBlockingQueue is backed by an OctreePriorityQueue, which is backed by an immutable Octree
* data structure, iterating over this queue actually is a thread safe operation which will iterate over
* the version of the items in the queue at the time the iterator was produced.
*/
override def iterator(): util.Iterator[(V3F, E)] = {
try {
readLock.lock()
queue.iterator()
} finally readLock.unlock()
}

override def size(): Int = {
try {
readLock.lock()
queue.size
} finally readLock.unlock()
}
}