Type aligned queue implementation

tqueue.scala

package typealigned

import scalaz.{Leibniz,Monad,Need}
import Leibniz.===

/** Typeclass for type-aligned sequences. */
trait Sequence[Q[_[_,_],_,_]] { self =>
  import syntax._
  def empty[C[_,_],a]: Q[C,a,a]
  def singleton[C[_,_],a,b](c: C[a,b]): Q[C,a,b]
  def uncons[C[_,_],a,b](s: Q[C,a,b]): ViewL[Q,C,a,b]
  def snoc[C[_,_],a,b,c](s: Q[C,a,b])(c: C[b,c]): Q[C,a,c]

  /** Linear time `append` operation, defined in terms of repeated `uncons`. */
  def linearAppend[C[_,_],a,b,c](s: Q[C,a,b])(s2: Q[C,b,c]): Q[C,a,c] = {
    val ht = uncons(s2)
    ht.handle { new ht.H[Q[C,a,c]] { def apply[x] =
      (h,t) => linearAppend(snoc(s)(h))(t.value)
    }} (e => align[C,a,b,c](s)(e))
  }
  def append[C[_,_],a,b,c](s: Q[C,a,b])(s2: Need[Q[C,b,c]]): Q[C,a,c]

  object viewL {
    def empty[C[_,_],a]: ViewL[Q,C,a,a] =
      ViewL.Empty()
    def cons[C[_,_],a,b,c](h: C[a,b], t: => Q[C,b,c]): ViewL[Q,C,a,c] =
      ViewL.Cons(h, Need(t))
  }

  /** Convert the output type `b` to `c`, given evidence that `b === c`. */
  def align[C[_,_],a,b,c](s: Q[C,a,b])(implicit e: b === c): Q[C,a,c] =
    e.subst[({ type f[b] = Q[C,a,b]})#f](s)

  /** Convert the input type `b` to `a`, given evidence that `b === c`. */
  def alignL[C[_,_],a,b,c](s: Q[C,b,c])(implicit e: b === a): Q[C,a,c] =
    e.subst[({ type f[a] = Q[C,a,c]})#f](s)

  /** Infix syntax for various operations. */
  object syntax {
    implicit class syntax[C[_,_],a,b](seq: Q[C,a,b]) {
      def snoc[c](c: C[b,c]): Q[C,a,c] =
        self.snoc[C,a,b,c](seq)(c)
      def align[c](implicit e: b === c): Q[C,a,c] =
        self.align[C,a,b,c](seq)(e)
      def alignL[a0](implicit e: a === a0): Q[C,a0,b] =
        self.alignL[C,a0,a,b](seq)(e)
      def uncons: ViewL[Q,C,a,b] =
        self.uncons(seq)
      def append[c](s2: => Q[C,b,c]): Q[C,a,c] =
        self.append(seq)(Need(s2))
      def ++[c](s2: => Q[C,b,c]): Q[C,a,c] =
        self.append(seq)(Need(s2))
    }

    implicit def symmEq[a,b](e: a === b): b === a =
      Leibniz.symm[Nothing,Any,a,b](e)

    implicit def cq[C[_,_],a,b](seq: Q[CQ.CQC[Q,C]#f,a,b]) =
      new syntax[CQ.CQC[Q,C]#f,a,b](seq)
  }
}

/**
 * A view of the left (cons) side of type-aligned sequences
 * of type `Q`. May be either empty, in which case we obtain
 * evidence that `a === b`, or a `Cons` which reveals
 * an (existential) intermediate type between `a` and `b`.
 */
trait ViewL[Q[_[_,_],_,_],C[_,_], a, b] {
  def handle[R](h: H[R])(r: (b === a) => R): R
  def asEmpty: Option[b === a]

  trait H[+R] {
    def apply[x]: (C[a,x], Need[Q[C,x,b]]) => R
  }
}

object ViewL {

  case class Empty[S[_[_,_],_,_], C[_,_], a]() extends ViewL[S,C,a,a] {
    def handle[R](h: H[R])(r: (a === a) => R) = r(Leibniz.refl[a])
    def asEmpty = Some(Leibniz.refl[a])
  }
  case class Cons[Q[_[_,_],_,_], C[_,_], a, b, c](
    head: C[a,b],
    tail: Need[Q[C,b,c]]) extends ViewL[Q, C, a, c] {
    def handle[R](h: H[R])(r: (c === a) => R) = h.apply[b](head, tail)
    def asEmpty = None
  }
}

import CQ.{CQC,CQP}

/**
 * Concatenable type-aligned queue, supports
 * append in same time complexity as underlying
 * queue's snoc operation.
 */
trait CQ[Q[_[_,_],_,_], C[_,_], a, b] {
  def handle[R](h: H[R])(r: (b === a) => R): R
  def asEmpty: Option[b === a]
  def snoc[c](c: C[b,c])(implicit Q: Sequence[Q]): CQ[Q,C,a,c] =
    CQ.instance[Q].snoc(this)(c)

  trait H[R] {
    def apply[x]: (C[a,x], Need[Q[({type f[x,y] = CQ[Q,C,x,y] })#f, x, b]]) => R
  }

  def align[c](e: Leibniz[_,_,b,c]): CQ[Q,C,a,c] =
    e.subst[({ type f[c] = CQ[Q,C,a,c]})#f](this)
}

object CQ {
  trait CQP[Q[_[_,_],_,_]] { type f[c[_,_],x,y] = CQ[Q,c,x,y] }
  trait CQC[Q[_[_,_],_,_],C[_,_]] { type f[x,y] = CQ[Q,C,x,y] }

  implicit def instance[Q[_[_,_],_,_]](implicit Q: Sequence[Q]):
  Sequence[CQP[Q]#f] =
    new Sequence[CQP[Q]#f] {

    import Q.syntax._

    def empty[C[_,_],a]: CQ[Q,C,a,a] =
      C0()

    def singleton[C[_,_],a,b](c: C[a,b]): CQ[Q,C,a,b] =
      CN(c, Need(Q.empty[CQC[Q,C]#f,b]))

    def uncons[C[_,_],a,b](s: CQ[Q,C,a,b]): ViewL[CQP[Q]#f,C,a,b] =
      s.handle { new s.H[ViewL[CQP[Q]#f,C,a,b]] { def apply[x] =
        (h,t) => viewL.cons(h, linkAll(t.value))
      }} (e => ViewL.Empty().asInstanceOf[ViewL[CQP[Q]#f,C,a,b]])

    def linkAll[C[_,_],a,b](q: Q[CQC[Q,C]#f,a,b]): CQ[Q,C,a,b] = {
      val ht = q.uncons
      ht.handle { new ht.H[CQ[Q,C,a,b]] { def apply[x] =
        (h,t) =>
          h.handle {
            new h.H[CQ[Q,C,a,b]] { def apply[y] =
              (x,q) => {
                val tl = linkAll(t.value)
                tl.asEmpty match {
                  case None =>
                    CN(x, Need(q.value snoc tl))
                  case Some(e) =>
                    CN(x, Need(q.value.align[b](e)))
                }
              }
            }
          } (_ => sys.error("unpossible!"))
        }
      } { e => align[C,a,a,b](empty[C,a])(e) }
    }

    def snoc[C[_,_],a,b,c](s: CQ[Q,C,a,b])(c: C[b,c]): CQ[Q,C,a,c] =
      append(s)(Need(singleton(c)))

    def append[C[_,_],a,b,c](s: CQ[Q,C,a,b])(
                             s2: Need[CQ[Q,C,b,c]]): CQ[Q,C,a,c] =
      s.handle { new s.H[CQ[Q,C,a,c]] { def apply[x] =
        (h,t) =>
          s2.value.asEmpty match {
            case None => CN(h, Need(t.value snoc s2.value))
            case Some(e) => s.align[c](e)
          }
      }} (e => alignL(s2.value)(e))
  }

  case class C0[Q[_[_,_],_,_], C[_,_], x]()
      extends CQ[Q,C,x,x] {
    def handle[R](h: H[R])(r: (x === x) => R): R = r(Leibniz.refl[x])
    def asEmpty = Some(Leibniz.refl[x])
  }

  case class CN[Q[_[_,_],_,_], C[_,_], x, y, z](
    head: C[x,y],
    tail: Need[Q[({type f[x,y] = CQ[Q,C,x,y] })#f, y, z]])
    extends CQ[Q,C,x,z] {
    def handle[R](h: H[R])(r: (z === x) => R): R = h.apply[y](head, tail)
    def asEmpty = None
  }
}

/**
 * Type-aligned queue supporting `snoc` and `uncons`
 * in worst-case `O(1)`.
 */
trait Q[C[_,_],x,y] {
  def snoc[z](f: C[y,z]): Q[C,x,z]
  def uncons: ViewL[Q,C,x,y]
}

object Q {
  import B._

  implicit def instance: Sequence[Q] = new Sequence[Q] {
    def append[C[_, _], a, b, c](s: Q[C,a,b])(s2: Need[Q[C,b,c]]): Q[C,a,c] = linearAppend(s)(s2.value)
    def empty[C[_, _], a]: Q[C,a,a] = Q0()
    def singleton[C[_, _], a, b](c: C[a,b]): Q[C,a,b] = Q1(c)
    def snoc[C[_, _], a, b, c](s: Q[C,a,b])(c: C[b,c]): Q[C,a,c] = s.snoc(c)
    def uncons[C[_, _], a, b](s: Q[C,a,b]): ViewL[Q,C,a,b] = s.uncons
  }

  implicit class SnocB[C[_,_],x,y](q: Q[P.PC[C]#f,x,y]) {
    def snocB[z](r: B[C,y,z]): Q[C,x,z] = {
      val qv = q.uncons
      val h = new qv.H[Q[C,x,z]] { def apply[zz] =
        (l,m) => QN(B2(l), m, r)
      }
      qv.handle(h) (
        ev => ev.subst[({type f[y] = Q[C,y,z]})#f](r.toQueue)
      )
    }
  }

  case class Q0[C[_,_],x]() extends Q[C,x,x] {
    def snoc[z](f: C[x,z]): Q[C,x,z] = Q1(f)
    def uncons: ViewL[Q,C,x,x] = ViewL.Empty()
  }
  case class Q1[C[_,_],x,y](f: C[x,y]) extends Q[C,x,y] {
    def snoc[z](g: C[y,z]): Q[C,x,z] =
      qN(B1(f), Q0[P.PC[C]#f,y](), B1(g))
    def uncons: ViewL[Q,C,x,y] =
      ViewL.Cons[Q,C,x,y,y](f, Need(Q0()))
  }

  def qN[C[_,_],a,b,x,y](
    left: B[C,a,x],
    middle: => Q[P.PC[C]#f, x, y],
    right: B[C, y, b]): Q[C,a,b] = {
    QN(left, Monad[Need].pure(middle), right)
  }

  case class QN[C[_,_],a,b,x,y](
    left: B[C,a,x],
    middle: Need[Q[P.PC[C]#f, x, y]],
    right: B[C,y,b])
    extends Q[C,a,b] {
      def snoc[c](g: C[b,c]): Q[C,a,c] =
        (right ~> g) match {
          case Left(r) => qN(left, middle.value.snoc(r), B1(g))
          case Right(right) => QN(left, middle, right)
        }
      def uncons: ViewL[Q,C,a,b] = left match {
        case B2(P.Compose(a,b)) => ViewL.Cons(a, Need(QN(B1(b), middle, right)))
        case B1(a) => ViewL.Cons(a, Need(middle.value.snocB(right)))
      }
  }
}

trait P[C[_,_],A,B] {
  def toQueue: Q[C,A,B]
}

object P {
  trait PC[C[_,_]] { type f[x,y] = P[C,x,y] }

  implicit class syntax[C[_,_],X,Y](f: C[X,Y]) {
    def ~>[Z](g: C[Y,Z]): P[C,X,Z] = Compose(f,g)
  }

  case class Compose[C[_,_],X,Y,Z](f: C[X,Y], g: C[Y,Z]) extends P[C,X,Z] {
    def toQueue: Q[C,X,Z] =
      Q.qN(B.B1(f), Q.Q0[PC[C]#f,Y](), B.B1(g))
  }
}

trait B[C[_,_],a,b] {
  def ~>[c](h: C[b,c]): Either[P[C,a,b], B[C,a,c]]
  def toQueue: Q[C,a,b]
}

object B {
  import P.syntax

  case class B1[C[_,_],a,b](f: C[a,b]) extends B[C,a,b] {
    def ~>[c](h: C[b,c]) = Right(B2(f ~> h))
    def toQueue = Q.Q1(f)
  }
  case class B2[C[_,_],a,b](p: P[C,a,b]) extends B[C,a,b] {
    def ~>[c](h: C[b,c]) = Left(p)
    def toQueue = p.toQueue
  }
}

In case you're interested, I started working on a port of the various type-aligned sequences described in the paper in scala tseq. Your gist here helped me with the representation quite a bit, especially the insight on using the Leibniz proof that the index types were equal in the empty case when folding the various sequences!

Anyhow, I've focused on implementing what was necessary to modify Free from scalaz, as well as just working to understand the paper, but plan on adding the rest of the sequences as time permits.