Github user hqzizania commented on the issue:
https://github.com/apache/spark/pull/13891
code for testing
```
def run(rank: Int, a:Int) = {
println(s"blas.getclass() = ${blas.getClass.toString} on process $rank")
val m = 1 << a
val n = 1 << a - 1
val stack = 1 << a - 2
val matrix = new Array[Array[Float]](m).map { x =>
val y = new Array[Float](n)
y.map(a => Random.nextFloat())
}
val bVector = new Array[Double](m).map(x => Random.nextDouble())
val ls = new NormalEquation(n)
for (u <- 0 to 3) {
ls.reset()
val t0 = System.nanoTime()
for (i <- 0 until m)
ls.add(matrix(i), bVector(i))
val t1 = System.nanoTime()
println("nostack Elapsed time: " + (t1 - t0) / 1000000 + s"ms on
process $rank")
ls.reset()
val t2 = System.nanoTime()
var i = 0
while (i < m) {
val matrixBuffer = mutable.ArrayBuilder.make[Double]
val bBuffer = mutable.ArrayBuilder.make[Double]
for (s <- 0 until stack) {
for (j <- 0 until n) {
matrixBuffer += matrix(i + s)(j)
}
bBuffer += bVector(i + s)
}
i += stack
ls.addStack(matrixBuffer.result(), bBuffer.result(), stack)
}
val t3 = System.nanoTime()
println("stack Elapsed time: " + (t3 - t2) / 1000000 + s"ms on
process $rank")
}
}
class NormalEquation(val k: Int) extends Serializable {
/** Number of entries in the upper triangular part of a k-by-k matrix.
*/
val triK = k * (k + 1) / 2
/** A^T^ * A */
val ata = new Array[Double](triK)
/** A^T^ * b */
val atb = new Array[Double](k)
private val da = new Array[Double](k)
private val ata2 = new Array[Double](k * k)
private val upper = "U"
private def copyToDouble(a: Array[Float]): Unit = {
var i = 0
while (i < k) {
da(i) = a(i)
i += 1
}
}
private def copyToTri(): Unit = {
var ii = 0
for(i <- 0 until k)
for(j <- 0 to i) {
ata(ii) += ata2(i * k + j)
ata2(i * k + j) = 0
ii += 1
}
}
/** Adds an observation. */
def add(a: Array[Float], b: Double, c: Double = 1.0): this.type = {
require(c >= 0.0)
require(a.length == k)
copyToDouble(a)
blas.dspr(upper, k, c, da, 1, ata)
if (b != 0.0) {
blas.daxpy(k, c * b, da, 1, atb, 1)
}
this
}
/** Adds a stack of observations. */
def addStack(a: Array[Double], b: Array[Double], n: Int): this.type = {
require(a.length == n * k)
blas.dsyrk(upper, "N", k, n, 1.0, a, k, 1.0, ata2, k)
copyToTri()
blas.dgemv("N", k, n, 1.0, a, k, b, 1, 1.0, atb, 1)
this
}
/** Merges another normal equation object. */
def merge(other: NormalEquation): this.type = {
require(other.k == k)
blas.daxpy(ata.length, 1.0, other.ata, 1, ata, 1)
blas.daxpy(atb.length, 1.0, other.atb, 1, atb, 1)
this
}
/** Resets everything to zero, which should be called after each solve.
*/
def reset(): Unit = {
ju.Arrays.fill(ata, 0.0)
ju.Arrays.fill(ata2, 0.0)
ju.Arrays.fill(atb, 0.0)
}
}
```
results:
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at [email protected] or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
