Как найти наиболее эффективные прямоугольные области с одинаковым значением для данного размера в матрице?

Question

Моя проблема очень проста, но я пока не нашел эффективной реализации.

Предположим, что существует матрица А как это:

0 0 0 0 0 0 0 
4 4 2 2 2 0 0 
4 4 2 2 2 0 0 
0 0 2 2 2 1 1 
0 0 0 0 0 1 1 

Теперь я хочу, чтобы найти все стартовые позиции прямоугольных областей в этой матрице, которые имеют заданный размер. Область является подмножеством A, где все числа одинаковы.

Предположим, ширина = 2 и высота = 3. Есть 3 области, которые имеют этот размер:

2 2 2 2 0 0 
2 2 2 2 0 0 
2 2 2 2 0 0 

В результате вызова функции будет представлять собой список стартовых позиций (х, у, начиная с 0) из этих областей.

List((2,1),(3,1),(5,0)) 

Ниже приводится моя текущая реализация. «Районы» здесь называются «поверхностями».

case class Dimension2D(width: Int, height: Int) 
case class Position2D(x: Int, y: Int) 

def findFlatSurfaces(matrix: Array[Array[Int]], surfaceSize: Dimension2D): List[Position2D] = { 

    val matrixWidth = matrix.length 
    val matrixHeight = matrix(0).length 
    var resultPositions: List[Position2D] = Nil 

    for (y <- 0 to matrixHeight - surfaceSize.height) { 
     var x = 0 
     while (x <= matrixWidth - surfaceSize.width) { 
      val topLeft = matrix(x)(y) 
      val topRight = matrix(x + surfaceSize.width - 1)(y) 
      val bottomLeft = matrix(x)(y + surfaceSize.height - 1) 
      val bottomRight = matrix(x + surfaceSize.width - 1)(y + surfaceSize.height - 1) 
      // investigate further if corners are equal 
      if (topLeft == bottomLeft && topLeft == topRight && topLeft == bottomRight) { 
       breakable { 
        for (sx <- x until x + surfaceSize.width; 
         sy <- y until y + surfaceSize.height) { 
         if (matrix(sx)(sy) != topLeft) { 
          x = if (x == sx) sx + 1 else sx 
          break 
         } 
        } 
        // found one!  
        resultPositions ::= Position2D(x, y) 
        x += 1 
       } 
      } else if (topRight != bottomRight) { 
       // can skip x a bit as there won't be a valid match in current row in this area 
       x += surfaceSize.width 
      } else { 
       x += 1 
      } 
     } 
    } 
    return resultPositions 
} 

Я уже пытался включить в него некоторые оптимизации, но я уверен, что есть гораздо лучшие решения. Существует ли для этого функция matlab, которую я мог бы переносить? Мне также интересно, имеет ли эта проблема свое имя, поскольку я точно не знал, для чего нужно google.

Спасибо, что подумали! Я рад видеть ваши предложения или решения :)

EDIT: Размеры матрицы в моем диапазоне применения от 300x300 до 3000x3000 приблизительно. Кроме того, алгоритм будет называться только для той же матрицы. Причина в том, что матрица всегда будет изменена впоследствии (примерно 1-20% от нее).

РЕЗУЛЬТАТЫ

Я реализовал алгоритмы Кевина, Никита и Даниил и протестировал их в своей среде приложений, т.е. нет изолированных синтетических тестов здесь, но особое внимание было уделено интеграцией всех алгоритмов в наиболее производительном способе, который был особенно важно для подхода Кевина, поскольку он использует дженерики (см. ниже).

Во-первых, необработанные результаты, используя Scala 2.8 и jdk 1.6.0_23. Алгоритмы выполнялись несколько сотен раз как часть решения конкретной задачи. «Длительность» означает общее время, необходимое для завершения алгоритма приложения (конечно, без запуска jvm и т. Д.). Моя машина представляет собой Core 2 Duo 2,8 ГГц с 2 ядрами и 2gig памяти, -Xmx800M были переданы JVM.

ВАЖНОЕ ЗАМЕЧАНИЕ: Я думаю, что моя эталонная установка не очень справедлива для распараллеливаемых алгоритмов, подобных тому, что у Даниила. Это связано с тем, что приложение уже вычисляет многопоточность. Таким образом, результаты здесь, вероятно, показывают только эквивалентную однопоточную скорость.

Размер матрицы 233x587:

    duration | JVM memory | avg CPU utilization 
original O(n^4) | 3000s  30M   100% 
original/-server| 840s  270M   100% 
Nikita O(n^2) | 5-6s  34M   70-80% 
Nikita/-server | 1-2s  300M   100% 
Kevin/-server | 7400s  800M   96-98% 
Kevin/-server** | 4900s  800M   96-99% 
Daniel/-server | 240s  360M   96-99% 

** с @specialized, чтобы сделать generics faster избегая Erasure

отсутствует Тип

размер матрицы 2000x3000:

    duration | JVM memory | avg CPU utilization 
original O(n^4) | too long 100M   100% 
Nikita O(n^2) | 150s  760M   70% 
Nikita/-server | 295s (!) 780M   100% 
Kevin/-server | too long, didn't try 

Во-первых, небольшую заметку на память , Опция -server JVM использует значительно больше памяти в интересах большего количества оптимизаций и, как правило, более быстрого выполнения. Как видно из второй таблицы, алгоритм Никиты работает медленнее с опцией -server, что, очевидно, связано с ограничением памяти. Я предполагаю, что это также замедляет алгоритм Кевина даже для маленькой матрицы, так как функциональный подход все равно использует гораздо больше памяти. Чтобы устранить фактор памяти, я также попробовал его один раз с матрицей 50x50, а затем Кевин взял 5 секунд и 0 секунд на Nikita (ну, почти 0). Поэтому в любом случае он все еще медленнее, а не только из-за памяти.

Как видно из цифр, я, очевидно, буду использовать алгоритм Никиты, потому что это чертовски быстро, и это абсолютно необходимо в моем случае. Он также может быть легко распараллелен, как указал Даниил. Единственным недостатком является то, что на самом деле это не scala-way.

В настоящий момент алгоритм Кевина, вероятно, в целом слишком сложный и, следовательно, медленный, но я уверен, что возможны дополнительные оптимизации (см. Последние комментарии в его ответе).

С целью прямого преобразования алгоритма Никиты в функциональный стиль Даниэль придумал решение, которое уже довольно быстро, и, по его словам, будет даже быстрее, если он сможет использовать scanRight (см. Последние комментарии в его ответе).

Что дальше?

С технологической стороны: ожидание Scala 2.9, ScalaCL и выполнение синтетических тестов для получения исходных скоростей.

Моя цель во всем этом состоит в том, чтобы иметь функциональный код, НО только если он не жертвует слишком большой скоростью.

Выбор ответа:

Что касается выбора ответа, я хотел бы отметить алгоритмы Никиту и Даниила как ответы, но я должен выбрать один. Название моего вопроса включало «наиболее эффективно», а другое - самое быстрое в императиве, а другое - в функциональном стиле. Хотя этот вопрос отмечен Scala, я выбрал императивный алгоритм Никиты, поскольку 2s против 240s по-прежнему слишком большой разницы для меня. Я уверен, что разницу можно еще немного оттолкнуть, какие-нибудь идеи?

Итак, спасибо вам всем очень много! Хотя я не буду использовать функциональные алгоритмы еще, у меня появилось много новых знаний о Scala, и я думаю, что я медленно понимаю все функциональные сумасшествия и его потенциал. (конечно, даже не делая много функционального программирования, Scala гораздо приятнее, чем Java ...это еще одна причина, чтобы узнать его)

Answer 1

Вы можете сделать это в O(n^2) относительно легко.
Во-первых, некоторые предварительные вычисления. Для каждой ячейки в матрице подсчитайте, сколько последовательных ячеек под ней имеют одинаковое число.
Для примера, в результате чего матрица a (не мог придумать лучшего названия: /) будет выглядеть следующим образом

0 0 0 0 0 2 2 
1 1 2 2 2 1 1 
0 0 1 1 1 0 0 
1 1 0 0 0 1 1 
0 0 0 0 0 0 0 

Он может быть получен в O(n^2) легко.

Теперь, для каждой строки i, давайте найдем все прямоугольники с верхней стороной в строке i (и нижняя сторона в ряду i + height - 1).
Вот иллюстрация для i = 1

0 0 0 0 0 0 0 
------------- 
4 4 2 2 2 0 0 
4 4 2 2 2 0 0 
0 0 2 2 2 1 1 
------------- 
0 0 0 0 0 1 1 

Теперь, основная идея

int current_width = 0; 
for (int j = 0; j < matrix.width; ++j) { 
    if (a[i][j] < height - 1) { 
     // this column has different numbers in it, no game 
     current_width = 0; 
     continue; 
    } 

    if (current_width > 0) { 
     // this column should consist of the same numbers as the one before 
     if (matrix[i][j] != matrix[i][j - 1]) { 
      current_width = 1; // start streak anew, from the current column 
      continue; 
     } 
    } 

    ++current_width; 
    if (current_width >= width) { 
     // we've found a rectangle! 
    } 
} 

В приведенном выше примере (i = 1) current_width после каждой итерации будет 0, 0, 1, 2, 3, 0, 0.

Теперь нам нужно перебрать все возможные i, и у нас есть решение.

Answer 2

Во-первых, несколько вспомогательных функций:

//count the number of elements matching the head 
def runLength[T](xs:List[T]) = xs.takeWhile(_ == xs.head).size 

//pair each element with the number of subsequent occurrences 
def runLengths[T](row:List[T]) : List[(T,Int)] = row match { 
    case Nil => Nil 
    case h :: t => (h, runLength(row)) :: runLengths(t) 
} 
//should be optimised for tail-call, but easier to understand this way 

//sample input: 1,1,2,2,2,3,4,4,4,4,5,5,6 
//output: (1,2), (1,1), (2,3), (2,2), (2,1), (3,1), (4,4), (4,3), (4,2), (4,1), (5,2), (5,1), (6,1) 

Это может быть использовано против каждой строки в сетке:

val grid = List(
    List(0,0,0,0), 
    List(0,1,1,0), 
    List(0,1,1,0), 
    List(0,0,0,0)) 

val stage1 = grid map runLengths 
// returns stage1: List[List[(Int, Int)]] = 
// 0,4 0,3 0,2 0,1 
// 0,1 1,2 1,1 0,1 
// 0,1 1,2 1,1 0,1 
// 0,4 0,3 0,2 0,1 

Затем, сделав горизонтальный , строки, мы теперь выполняем точно такую ​​же операцию над столбцами. Это использует метод transpose, доступный в стандартной библиотеке коллекций Scala, для обмена строк < -> столбцов в соответствии с операцией математического матрица с тем же именем. Мы также переносим назад, как только это будет сделано.

val stage2 = (stage1.transpose map runLengths).transpose 
// returns stage2: List[List[((Int, Int), Int)]] = 
// (0,4),1 (0,3),1 (0,2),1 (0,1),4 
// (0,1),2 (1,2),2 (1,1),2 (0,1),3 
// (0,1),1 (1,2),1 (1,1),1 (0,1),2 
// (0,4),1 (0,3),1 (0,2),1 (0,1),1 

Что это значит? Принимая один элемент: (1,2),2, это означает, что эта ячейка содержит значение 1 и сканирование вправо, что в соседней ячейке есть две соседние ячейки, содержащие 1. При сканировании существуют две соседние ячейки с тем же свойством, что и значение 1 и имеющие одинаковое количество равных значений справа.

Это немного яснее после уборки, преобразование вложенных кортежей вида ((a,b),c) к (a,(b,c)):

val stage3 = stage2 map { _.map {case ((a,b),c) => a->(b,c) } } 
//returns stage3: List[List[(Int, (Int, Int))]] = 
// 0,(4,1) 0,(3,1) 0,(2,1) 0,(1,4) 
// 0,(1,2) 1,(2,2) 1,(1,2) 0,(1,3) 
// 0,(1,1) 1,(2,1) 1,(1,1) 0,(1,2) 
// 0,(4,1) 0,(3,1) 0,(2,1) 0,(1,1) 

Где 1,(2,2) относится к клетке, содержащей значение 1, и, будучи в верхнем левом углу 2x2 прямоугольника одинаковозначных ячеек.

Отсюда тривиальные пятна прямоугольников одинакового размера. Хотя требуется немного больше работы, если вы также хотите исключить области, которые являются подмножеством большего прямоугольника.

UPDATE: Как написано, алгоритм работает не так, как в ячейке at (0,0), которая принадлежит двум различным прямоугольникам (1x4 и 4x1). Если необходимо, это также разрешимо с использованием тех же методов. (сделать один проход, используя map/transpose/map/transpose, а второй с транспозицией/отображением/транспозицией/отображением, затем застегнуть молнию и сгладить результаты).

Было бы также нужно модифицировать, если вход может содержать примыкающие прямоугольники, содержащие клетки одного и того же значения, такие как:

0 0 0 0 0 0 0 0 
0 0 1 1 1 0 0 0 
0 0 1 1 1 0 0 0 
0 0 1 1 1 1 1 0 
0 0 1 1 1 1 1 0 
0 0 1 1 1 1 1 0 
0 0 0 0 0 0 0 0 

Собираем все вместе, и очистки немного:

type Grid[T] = List[List[T]] 

def runLengths[T](row:List[T]) : List[(T,Int)] = row match { 
    case Nil => Nil 
    case h :: t => (h -> row.takeWhile(_ == h).size) :: runLengths(t) 
} 

def findRectangles[T](grid: Grid[T]) = { 
    val step1 = (grid map runLengths) 
    val step2 = (step1.transpose map runLengths).transpose 
    step2 map { _ map { case ((a,b),c) => (a,(b,c)) } } 
} 

UPDATE2

Удерживать на ваши шляпы, это большой один ...

Прежде чем написать одну строку новой функциональности, мы сначала немного реорганизуем, потянув некоторые из методов в классы Ops с неявными преобразованиями, поэтому их можно использовать так, как если бы они были методами, определенными в базовых типах коллекций:

import annotation.tailrec 

class RowOps[T](row: List[T]) { 
    def withRunLengths[U](func: (T,Int)=>U) : List[U] = { 
    @tailrec def recurse(row:List[T], acc:List[U]): List[U] = row match { 
     case Nil => acc 
     case head :: tail => 
     recurse(
      tail, 
      func(head, row.takeWhile(head==).size) :: acc) 
    } 
    recurse(row, Nil).reverse 
    } 

    def mapRange(start: Int, len: Int)(func: T=>T) = 
    row.splitAt(start) match { 
     case (l,r) => l ::: r.take(len).map(func) ::: r.drop(len) 
    } 
} 

implicit def rowToOps[T](row: List[T]) = new RowOps(row) 

Это добавляет метод withRunLengths к спискам. Одна из отличительных отличий здесь заключается в том, что вместо того, чтобы возвращать пары из (value, length), метод принимает в качестве параметра функцию для создания выходного значения для каждой такой пары.Это будет полезно позже ...

type Grid[T] = List[List[T]] 

class GridOps[T](grid: Grid[T]) { 
    def deepZip[U](other: Grid[U]) = (grid zip other) map { case (g,o) => g zip o} 
    def deepMap[U](f: (T)=>U) = grid map { _ map f} 
    def mapCols[U](f: List[T]=>List[U]) = (grid.transpose map f).transpose 
    def height = grid.size 
    def width = grid.head.size 
    def coords = List.tabulate(height,width){ case (y,x) => (x,y) } 
    def zipWithCoords = deepZip(coords) 
    def deepMapRange(x: Int, y: Int, w: Int, h: Int)(func: T=>T) = 
    grid mapRange (y,h){ _.mapRange(x,w)(func) } 
} 

implicit def gridToOps[T](grid: Grid[T]) = new GridOps(grid) 

Здесь не должно быть никаких сюрпризов. Методы deepXXX избегают писать конструкции вида list map { _ map { ... } }. tabulate также может быть для вас новичком, но его цель, надеюсь, очевидна из-за использования.

С их помощью становится тривиальным определить функции для нахождения горизонтальных и вертикальных тиражей по всей сетке:

def withRowRunLengths[T,U](grid: Grid[T])(func: (T,Int)=>U) = 
    grid map { _.withRunLengths(func) } 

def withColRunLengths[T,U](grid: Grid[T])(func: (T,Int)=>U) = 
    grid mapCols { _.withRunLengths(func) } 

Почему 2 блока параметров, а не один? Я скоро объясню.

I может определил их как методы в GridOps, но они не подходят для общего использования. Не стесняйтесь согласиться со мной здесь :)

Далее определимся ввод ...

def parseIntGrid(rows: String*): Grid[Int] = 
    rows.toList map { _ map {_.toString.toInt} toList } 

val input: Grid[Int] = parseIntGrid("0000","0110","0110","0000") 

... еще один полезный тип помощника ...

case class Rect(w: Int, h: Int) 
object Rect { def empty = Rect(0,0) } 

в качестве альтернативы кортеж, это действительно помогает при отладке. Глубоко вложенные круглые скобки: не Легко на глаза. (! жаль лисповские вентиляторы)

... и использовать функции, определенные выше:

val stage1w = withRowRunLengths(input) { 
    case (cell,width) => (cell,width) 
} 

val stage2w = withColRunLengths(stage1w) { 
    case ((cell,width),height) => Rect(width,height) 
} 


val stage1t = withColRunLengths(input) { 
case (cell,height) => (cell,height) 
} 

val stage2t = withRowRunLengths(stage1t) { 
    case ((cell,height),width) => Rect(width,height) 
} 

Все из вышеуказанных блоков должны быть однострочечники, но я переформатировать для StackOverflow.

Выходы на этом этапе являются только сетками прямоугольников, я намеренно отказался от упоминания фактического значения, которое состоит из прямоугольника. Это абсолютно нормально, достаточно легко найти его координаты в сетке, и мы будем рекомбинировать данные всего за короткое время.

Помните, что RowOps#withRunLengths принимает функцию как параметр? Ну, вот где он используется. case (cell,width) => (cell,width) - фактически функция, которая принимает значение ячейки и длину выполнения (вызывая их cell и width), а затем возвращает (cell,width) Tuple.

Вот почему я использовал два блока параметров при определении функций, поэтому второй параметр может быть передан в {скобки} и делает все это приятным и DSL-подобным.

Другого очень важный принцип иллюстрируется здесь в том, что тип inferencer работает на блоках параметров в последовательности, так что для этого (помните?):

def withRowRunLengths[T,U](grid: Grid[T])(func: (T,Int)=>U) 

Тип T будет определяться поставляемой сеткой , Затем компилятор знает тип ввода для функции, предоставленной в качестве второго аргумента, - Int в этом случае, поскольку первым параметром был Grid[Int] - вот почему я смог написать case (cell,width) => (cell,width) и не должен явно указывать в любом месте, где cell и width являются целыми числами.При втором использовании предоставляется Grid[(Int,Int)], это соответствует закрытию case ((cell,width),height) => Rect(width,height), и снова это просто работает.

Если что закрытие содержалось ничего, что не будет работать для базового типа сетки, то компилятор бы жаловался, это то, что безопасность типа все о ...

Рассчитав все возможные прямоугольники , остается только собрать данные и представить их в формате, который более практичен для анализа. Поскольку гнездования на этом этапе может получить очень грязный, я определил другой тип данных:

case class Cell[T](
    value: T, 
    coords: (Int,Int) = (0,0), 
    widest: Rect = Rect.empty, 
    tallest: Rect = Rect.empty 
) 

Ничего особенного, просто случай класса с именованными параметрами/по умолчанию. Я также рад, что я предусмотрительно определить Rect.empty выше :)

Теперь смешайте 4 наборов данных (входные Vals, COORDS, широкие прямоугольники высоких прямоугольников), постепенно складываются в клетки, размешать осторожно, и служить:

val cellsWithCoords = input.zipWithCoords deepMap { 
    case (v,(x,y)) => Cell(value=v, coords=(x,y)) 
} 

val cellsWithWidest = cellsWithCoords deepZip stage2w deepMap { 
    case (cell,rect) => cell.copy(widest=rect) 
} 

val cellsWithWidestAndTallest = cellsWithWidest deepZip stage2t deepMap { 
    case (cell,rect) => cell.copy(tallest=rect) 
} 

val results = (cellsWithWidestAndTallest deepMap { 
    case Cell(value, coords, widest, tallest) => 
    List((widest, value, coords), (tallest, value, coords)) 
    } 
).flatten.flatten 

последний этап интересен там, он преобразует каждую ячейку в список размера 2 (прямоугольник, значения, координаты) кортежами (размер 2, потому что у нас есть и самая широкие и самые высокие прямоугольники для каждой ячейки). Вызвав два раза сглаживанием, он возвращает List[List[List[_]]] обратно к одному List. Нет необходимости сохранять 2D-структуру больше, поскольку необходимые координаты уже встроены в результаты.

Voila!

Теперь вам решать, что вы сейчас делаете с этим списком. Следующим этапом, вероятно, является некоторая форма сортировки и удаления дубликатов ...

Answer 3

Я буду играть адвоката дьявола здесь. Я покажу Nikita's точный алгоритм, написанный в функциональном стиле. Я также распараллеливаю его, просто чтобы показать, что это можно сделать.

Во-первых, вычисление последовательных ячеек с таким же числом под ячейкой. Я сделал небольшое изменение, вернув все значения плюс один по сравнению с предлагаемым выходом Никиты, чтобы избежать - 1 в какой-то другой части кода.

def computeHeights(column: Array[Int]) = (
    column 
    .reverse 
    .sliding(2) 
    .map(pair => pair(0) == pair(1)) 
    .foldLeft(List(1)) ( 
     (list, flag) => (if (flag) list.head + 1 else 1) :: list 
    ) 
) 

Я предпочел бы использовать .view перед разворотом, но это не работает с текущей версии Scala. Если бы это было так, это спасло бы повторяющиеся массивные создания, которые должны были бы ускорить кодекс, для определения местоположения памяти и полосы пропускания, если нет другого.

Теперь все столбцы, в то же время:

import scala.actors.Futures.future 

def getGridHeights(grid: Array[Array[Int]]) = (
    grid 
    .transpose 
    .map(column => future(computeHeights(column))) 
    .map(_()) 
    .toList 
    .transpose 
) 

Я не уверен, если накладные расходы распараллеливание окупится здесь или нет, но это первый алгоритм на переполнение стека, где он на самом деле имеет шанс, учитывая нетривиальное усилие при вычислении столбца. Вот еще один способ записи его, используя запланированные 2.9 функции (она может работать на Scala 2.8.1 - не уверен, что:

def getGridHeights(grid: Array[Array[Int]]) = (
    grid 
    .transpose 
    .toParSeq 
    .map(computeHeights) 
    .toList 
    .transpose 
) 

Теперь мясо алгоритма Никите:

def computeWidths(height: Int, row: Array[Int], heightRow: List[Int]) = (
    row 
    .sliding(2) 
    .zip(heightRow.iterator) 
    .toSeq 
    .reverse 
    .foldLeft(List(1)) { case (widths @ (currWidth :: _), (Array(prev, curr), currHeight)) => 
     (
      if (currHeight >= height && currWidth > 0 && prev == curr) currWidth + 1 
      else 1 
     ) :: widths 
    } 
    .toArray 
) 

I Я использую сопоставление с образцом в этом фрагменте кода, хотя я заинтересован в его скорости, потому что со всеми скользящими и застегивающимися и складывающимися здесь есть две вещи. И, говоря об эффективности, я использую вместо этого Array от IndexedSeq, потому что Array - единственный тип в JVM, который не стирается, что приводит к значительно лучшей производительности с Int. И тогда, есть .toSeq, я тоже не очень доволен, из-за локальности памяти и пропускной способности.

Кроме того, я делаю это справа налево вместо Никиты слева направо, так что я могу найти верхний левый угол.

Однако это то же самое, что и код в ответе Никиты, помимо того факта, что я все еще добавляю ширину по сравнению с его кодом и не печатаю результат прямо здесь. Здесь есть куча вещей без ясности, но - row, heightRow, height ... Давайте посмотрим на этот код в контексте - и распараллеливаем! - получить общую картину.

def getGridWidths(height: Int, grid: Array[Array[Int]]) = (
    grid 
    .zip(getGridHeights(grid)) 
    .map { case (row, heightsRow) => future(computeWidths(height, row, heightsRow)) } 
    .map(_()) 
) 

И версия 2.9:

def getGridWidths(height: Int, grid: Array[Array[Int]]) = (
    grid 
    .toParSeq 
    .zip(getGridHeights(grid)) 
    .map { case (row, heightsRow) => computeWidths(height, row, heightsRow) } 
) 

И для Огап финала,

def findRectangles(height: Int, width: Int, grid: Array[Array[Int]]) = { 
    val gridWidths = getGridWidths(height, grid) 
    for { 
     y <- gridWidths.indices 
     x <- gridWidths(y).indices 
     if gridWidths(y)(x) >= width 
    } yield (x, y) 
} 

Так что ... Я не сомневаюсь, что императивный вариант алгоритма Никитского быстрее - он использует только Array, что намного быстрее с примитивами, чем любой другой тип, и это позволяет избежать массового создания временных коллекций - хотя Scala может здесь сделали. Кроме того, нет замыканий - насколько они помогают, они являются медленнее, чем код без замыканий. По крайней мере, пока JVM не вырастет, чтобы помочь им.

Кроме того, код Никиты можно легко распараллелить нитями - из всех вещей! - с небольшими неприятностями.

Но я хочу сказать, что код Никиты не особенно плох, потому что он использует массивы и изменяемую переменную здесь и там. Алгоритм полностью переводится в более функциональный стиль.

EDIT

Итак, я решил взять выстрел на создание эффективной функциональной версии. Это не совсем функционально, потому что я использую Iterator, который является изменяемым, но он достаточно близко. К сожалению, он не будет работать на Scala 2.8.1, потому что ему не хватает scanLeft на Iterator.

Здесь есть две другие неудачные вещи. Во-первых, я отказался от распараллеливания высот сетки, так как я не мог обойтись, имея хотя бы один transpose со всеми копиями коллекции, которые влекут за собой. Однако есть хотя бы одно копирование данных (см. toArray звонок, чтобы понять, где).

Кроме того, поскольку я работаю с Iterable, я теряю способность использовать параллельные коллекции. Я действительно задаюсь вопросом, нельзя ли улучшить код, если grid будет параллельным коллекцией параллельных коллекций с самого начала.

У меня нет подсказки, если это более эффективно, чем предыдущая версия. Это интересный вопрос ...

def getGridHeights(grid: Array[Array[Int]]) = (
    grid 
    .sliding(2) 
    .scanLeft(Array.fill(grid.head.size)(1)) { case (currHeightArray, Array(prevRow, nextRow)) => 
     (prevRow, nextRow, currHeightArray) 
     .zipped 
     .map { case (x, y, currHeight) => if (x == y) currHeight + 1 else 1 } 
    } 
) 

def computeWidths(height: Int, row: Array[Int], heightRow: Array[Int]) = (
    row 
    .sliding(2) 
    .map { case Array(x, y) => x == y } 
    .zip(heightRow.iterator) 
    .scanLeft(1) { case (currWidth , (isConsecutive, currHeight)) => 
     if (currHeight >= height && currWidth > 0 && isConsecutive) currWidth + 1 
     else 1 
    } 
    .toArray 
) 

import scala.actors.Futures.future 

def getGridWidths(height: Int, grid: Array[Array[Int]]) = (
    grid 
    .iterator 
    .zip(getGridHeights(grid)) 
    .map { case (row, heightsRow) => future(computeWidths(height, row, heightsRow)) } 
    .map(_()) 
    .toArray 
) 

def findRectangles(height: Int, width: Int, grid: Array[Array[Int]]) = { 
    val gridWidths = getGridWidths(height, grid) 
    for { 
     y <- gridWidths.indices 
     x <- gridWidths(y).indices 
     if gridWidths(y)(x) >= width 
    } yield (x - width + 1, y - height + 1) 
}