Lineares Lesen eines mehrdimensionalen Arrays, das dimensionalen Unterabschnitten folgt

Question

Ich habe eine API zum Lesen mehrdimensionaler Arrays, die einen Vektor von Bereichen zum Lesen von Unterrechtecken (oder Hypercubes) aus dem Backing-Array übergeben müssen. Ich möchte dieses Array "linear" lesen, alle Elemente in einer bestimmten Reihenfolge mit beliebigen Chunk-Größen. Somit besteht die Aufgabe darin, mit einem off und einem len die durch diesen Bereich abgedeckten Elemente in den kleinstmöglichen Satz von Hyperwürfeln zu übersetzen, d.h. die kleinste Anzahl von Lesebefehlen, die in der API ausgegeben werden.

Zum Beispiel können wir Indexvektoren für den Satz von Dimensionen geben einen linearen Index berechnen:

def calcIndices(off: Int, shape: Vector[Int]): Vector[Int] = { 
    val modsDivs = shape zip shape.scanRight(1)(_ * _).tail 
    modsDivs.map { case (mod, div) => 
    (off/div) % mod 
    } 
} 

Nehmen wir an, die Form ist dies, was ein Array mit Rang 4 und 120 Elemente in Summe:

val sz = Vector(2, 3, 4, 5) 
val num = sz.product // 120 

ein Dienstprogramm diese Indexvektoren für einen Bereich von linearen Offsets drucken:

def printIndices(off: Int, len: Int): Unit = 
    (off until (off + len)).map(calcIndices(_, sz)) 
    .map(_.mkString("[", ", ", "]")).foreach(println) 

Wir können alle diese Vektoren erzeugen:

printIndices(0, num) 

[0, 0, 0, 0] 
[0, 0, 0, 1] 
[0, 0, 0, 2] 
[0, 0, 0, 3] 
[0, 0, 0, 4] 
[0, 0, 1, 0] 
[0, 0, 1, 1] 
[0, 0, 1, 2] 
[0, 0, 1, 3] 
[0, 0, 1, 4] 
[0, 0, 2, 0] 
[0, 0, 2, 1] 
[0, 0, 2, 2] 
[0, 0, 2, 3] 
[0, 0, 2, 4] 
[0, 0, 3, 0] 
[0, 0, 3, 1] 
[0, 0, 3, 2] 
[0, 0, 3, 3] 
[0, 0, 3, 4] 
[0, 1, 0, 0] 
... 
[1, 2, 1, 4] 
[1, 2, 2, 0] 
[1, 2, 2, 1] 
[1, 2, 2, 2] 
[1, 2, 2, 3] 
[1, 2, 2, 4] 
[1, 2, 3, 0] 
[1, 2, 3, 1] 
[1, 2, 3, 2] 
[1, 2, 3, 3] 
[1, 2, 3, 4] 

auf ein Beispiel Brocken aussehen lassen, die gelesen werden soll, die ersten sechs Elemente:

val off1 = 0 
val len1 = 6 
printIndices(off1, len1) 

Ich werde schon die Ausgabe von Hand in Hyper partitionieren:

// first hypercube or read 
[0, 0, 0, 0] 
[0, 0, 0, 1] 
[0, 0, 0, 2] 
[0, 0, 0, 3] 
[0, 0, 0, 4] 

// second hypercube or read 
[0, 0, 1, 0] 

So ist die Aufgabe, ein Verfahren

zu definieren

def partition(shape: Vector[Int], off: Int, len: Int): List[Vector[Range]] 

, die die richtige Liste ausgibt und die kleinstmögliche Listengröße verwendet. So für off1 und len1, haben wir das erwartete Ergebnis:

val res1 = List(
    Vector(0 to 0, 0 to 0, 0 to 0, 0 to 4), 
    Vector(0 to 0, 0 to 0, 1 to 1, 0 to 0) 
) 

assert(res1.map(_.map(_.size).product).sum == len1) 

Ein zweites Beispiel, Elemente bei Indizes 6 bis 22, mit manueller Partitionierung gibt drei Hyper oder Lesebefehle:

val off2 = 6 
val len2 = 16 
printIndices(off2, len2) 

// first hypercube or read 
[0, 0, 1, 1] 
[0, 0, 1, 2] 
[0, 0, 1, 3] 
[0, 0, 1, 4] 

// second hypercube or read 
[0, 0, 2, 0] 
[0, 0, 2, 1] 
[0, 0, 2, 2] 
[0, 0, 2, 3] 
[0, 0, 2, 4] 
[0, 0, 3, 0] 
[0, 0, 3, 1] 
[0, 0, 3, 2] 
[0, 0, 3, 3] 
[0, 0, 3, 4] 

// third hypercube or read 
[0, 1, 0, 0] 
[0, 1, 0, 1] 

expected result: 

val res2 = List(
    Vector(0 to 0, 0 to 0, 1 to 1, 1 to 4), 
    Vector(0 to 0, 0 to 0, 2 to 3, 0 to 4), 
    Vector(0 to 0, 1 to 1, 0 to 0, 0 to 1) 
) 

assert(res2.map(_.map(_.size).product).sum == len2) 

Beachten Sie, dass für val off3 = 6; val len3 = 21 würden wir vier Lesungen benötigen.

Answer 1

Die Idee des folgenden Algorithmus ist wie folgt:

• ein Point-of-Interest (POI) ist die am weitesten links stehende Position bei dem zwei Index-Repräsentationen (zum Beispiel unterscheiden sich für [0, 0, 0, 1] und [0, 1, 0, 0] das POI 1)
• wir die ursprüngliche (Start, Stopp) linearen Indexbereich
• verwenden wir Bewegungen in zwei Richtungen rekursiv unterteilen, wird zuerst die konstante indem Start- und den Stopp durch eine spezielle „ceil“ abnehm ope Ration an dem Start, später durch den Anschlag konstant zu halten und an dem Anschlag
• für jeden Teilbereich des Start durch einen speziellen „Boden“ Betrieb zu erhöhen, berechnen wir die poi der Grenzen und wir berechnen „trunc“ die ceil oder auf dem Boden ist der Betrieb wie oben beschrieben
• wenn diese trunc Wert an seinem Eingang identisch ist, fügen wir die gesamte Region und zurück
• sonst Rekursion wir
• die spezielle „ceil“ Betrieb nimmt den vorherigen Startwert und erhöht das Element am Poi-Index und setzt die nachfolgenden Elemente auf Null ; z.B. für [0, 0, 1, 1] und poi = 2 wäre die ceil [0, 0, 2, 0]
• die spezielle "floor" -Operation nimmt den vorherigen Stop-Wert und Nullen die Elemente nach dem Poi-Index; z.B. für [0, 0, 1, 1] und poi = 2 würde der Boden [0, 0, 1, 0]

Hier meine Implementierung sein soll. Zunächst ein paar Hilfsfunktionen:

def calcIndices(off: Int, shape: Vector[Int]): Vector[Int] = { 
    val modsDivs = (shape, shape.scanRight(1)(_ * _).tail, shape.indices).zipped 
    modsDivs.map { case (mod, div, idx) => 
    val x = off/div 
    if (idx == 0) x else x % mod 
    } 
} 

def calcPOI(a: Vector[Int], b: Vector[Int], min: Int): Int = { 
    val res = (a.drop(min) zip b.drop(min)).indexWhere { case (ai,bi) => ai != bi } 
    if (res < 0) a.size else res + min 
} 

def zipToRange(a: Vector[Int], b: Vector[Int]): Vector[Range] = 
    (a, b).zipped.map { (ai, bi) => 
    require (ai <= bi) 
    ai to bi 
    } 

def calcOff(a: Vector[Int], shape: Vector[Int]): Int = { 
    val divs = shape.scanRight(1)(_ * _).tail 
    (a, divs).zipped.map(_ * _).sum 
} 

def indexTrunc(a: Vector[Int], poi: Int, inc: Boolean): Vector[Int] = 
    a.zipWithIndex.map { case (ai, i) => 
    if  (i < poi) ai 
    else if (i > poi) 0 
    else if (inc)  ai + 1 
    else    ai 
    } 

Danach wird der aktuelle Algorithmus:

def partition(shape: Vector[Int], off: Int, len: Int): List[Vector[Range]] = { 
    val rankM = shape.size - 1 

    def loop(start: Int, stop: Int, poiMin: Int, dir: Boolean, 
      res0: List[Vector[Range]]): List[Vector[Range]] = 
    if (start == stop) res0 else { 
     val last = stop - 1 
     val s0 = calcIndices(start, shape) 
     val s1 = calcIndices(stop , shape) 
     val s1m = calcIndices(last , shape) 
     val poi = calcPOI(s0, s1m, poiMin) 
     val ti = if (dir) s0 else s1 
     val to = if (dir) s1 else s0 
     val st = if (poi >= rankM) to else indexTrunc(ti, poi, inc = dir) 

     val trunc = calcOff(st, shape) 
     val split = trunc != (if (dir) stop else start) 

     if (split) { 
     if (dir) { 
      val res1 = loop(start, trunc, poiMin = poi+1, dir = true , res0 = res0) 
      loop   (trunc, stop , poiMin = 0 , dir = false, res0 = res1) 
     } else { 
      val s1tm = calcIndices(trunc - 1, shape) 
      val res1 = zipToRange(s0, s1tm) :: res0 
      loop   (trunc, stop , poiMin = poi+1, dir = false, res0 = res1) 
     } 
     } else { 
     zipToRange(s0, s1m) :: res0 
     } 
    } 

    loop(off, off + len, poiMin = 0, dir = true, res0 = Nil).reverse 
} 

Beispiele:

val sz = Vector(2, 3, 4, 5) 
partition(sz, 0, 6) 

// result: 
List(
    Vector(0 to 0, 0 to 0, 0 to 0, 0 to 4), // first hypercube 
    Vector(0 to 0, 0 to 0, 1 to 1, 0 to 0) // second hypercube 
) 

partition(sz, 6, 21) 

// result: 
List(
    Vector(0 to 0, 0 to 0, 1 to 1, 1 to 4), // first read 
    Vector(0 to 0, 0 to 0, 2 to 3, 0 to 4), // second read 
    Vector(0 to 0, 1 to 1, 0 to 0, 0 to 4), // third read 
    Vector(0 to 0, 1 to 1, 1 to 1, 0 to 1) // fourth read 
) 

Die maximale Anzahl der liest, wenn ich mich nicht irre, wäre 2 * rank.