Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Collections API: Use Futures #70

Merged
merged 2 commits into from
Sep 1, 2017
Merged

Collections API: Use Futures #70

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
0164f20
Use Futures instead of blocking operations
rajadain Aug 30, 2017
c37560f
Cleanup
rajadain Aug 30, 2017
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
38 changes: 24 additions & 14 deletions api/src/main/scala/Geoprocessing.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,6 +1,8 @@
package org.wikiwatershed.mmw.geoprocessing

import java.util.concurrent.atomic.{LongAdder, DoubleAdder}
import scala.concurrent.Future
import scala.concurrent.ExecutionContext.Implicits.global

import collection.concurrent.TrieMap

Expand All @@ -17,10 +19,13 @@ trait Geoprocessing extends Utils {
* @param input The InputData
* @return A histogram of results
*/
def getRasterGroupedCount(input: InputData): ResultInt = {
def getRasterGroupedCount(input: InputData): Future[ResultInt] = {
val aoi = createAOIFromInput(input)
val rasterLayers = cropRastersToAOI(input.rasters, input.zoom, aoi)
ResultInt(rasterGroupedCount(rasterLayers, aoi))
val futureLayers = cropRastersToAOI(input.rasters, input.zoom, aoi)

futureLayers.map { layers =>
ResultInt(rasterGroupedCount(layers, aoi))
}
}

/**
Expand All @@ -30,21 +35,23 @@ trait Geoprocessing extends Utils {
* @param input The InputData
* @return A histogram of results
*/
def getRasterGroupedAverage(input: InputData): ResultDouble = {
def getRasterGroupedAverage(input: InputData): Future[ResultDouble] = {
val aoi = createAOIFromInput(input)
val rasterLayers = cropRastersToAOI(input.rasters, input.zoom, aoi)
val futureLayers = cropRastersToAOI(input.rasters, input.zoom, aoi)
val targetLayer = input.targetRaster match {
case Some(targetRaster) =>
cropSingleRasterToAOI(targetRaster, input.zoom, aoi)
case None =>
throw new Exception("Request data missing required 'targetRaster'.")
}

val average =
if (rasterLayers.isEmpty) rasterAverage(targetLayer, aoi)
else rasterGroupedAverage(rasterLayers, targetLayer, aoi)
futureLayers.map { rasterLayers =>
val average =
if (rasterLayers.isEmpty) rasterAverage(targetLayer, aoi)
else rasterGroupedAverage(rasterLayers, targetLayer, aoi)

ResultDouble(average)
ResultDouble(average)
}
}

/**
Expand All @@ -53,9 +60,9 @@ trait Geoprocessing extends Utils {
* @param input The InputData
* @return A histogram of results
*/
def getRasterLinesJoin(input: InputData): ResultInt = {
def getRasterLinesJoin(input: InputData): Future[ResultInt] = {
val aoi = createAOIFromInput(input)
val rasterLayers = cropRastersToAOI(input.rasters, input.zoom, aoi)
val futureLayers = cropRastersToAOI(input.rasters, input.zoom, aoi)
val lines = input.vector match {
case Some(vector) =>
input.vectorCRS match {
Expand All @@ -67,7 +74,10 @@ trait Geoprocessing extends Utils {
case None =>
throw new Exception("Request data missing required 'vector'.")
}
ResultInt(rasterLinesJoin(rasterLayers, lines))

futureLayers.map { rasterLayers =>
ResultInt(rasterLinesJoin(rasterLayers, lines))
}
}

private case class TilePixel(key: SpatialKey, col: Int, row: Int)
Expand All @@ -85,7 +95,7 @@ trait Geoprocessing extends Utils {
lines: Seq[MultiLine]
): Map[String, Int] = {
val metadata = rasterLayers.head.metadata
var pixelGroups: TrieMap[(List[Int], TilePixel), Int] = TrieMap.empty
val pixelGroups: TrieMap[(List[Int], TilePixel), Int] = TrieMap.empty

joinCollectionLayers(rasterLayers).par
.foreach({ case (key, tiles) =>
Expand Down Expand Up @@ -213,7 +223,7 @@ trait Geoprocessing extends Utils {
// assume all the layouts are the same
val metadata = rasterLayers.head.metadata

var pixelGroups: TrieMap[List[Int], LongAdder] = TrieMap.empty
val pixelGroups: TrieMap[List[Int], LongAdder] = TrieMap.empty

joinCollectionLayers(rasterLayers).par
.foreach({ case (key, tiles) =>
Expand Down
11 changes: 7 additions & 4 deletions api/src/main/scala/Utils.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,12 +1,13 @@
package org.wikiwatershed.mmw.geoprocessing

import scala.concurrent.Future
import scala.concurrent.ExecutionContext.Implicits.global

import spray.json._
import spray.json.DefaultJsonProtocol._

import geotrellis.proj4.{CRS, ConusAlbers, LatLng, WebMercator}

import geotrellis.raster._
import geotrellis.raster.rasterize._
import geotrellis.vector._
import geotrellis.vector.io._
import geotrellis.spark._
Expand All @@ -32,8 +33,10 @@ trait Utils {
rasterIds: List[String],
zoom: Int,
aoi: MultiPolygon
): Seq[TileLayerCollection[SpatialKey]] =
rasterIds.map { str => cropSingleRasterToAOI(str, zoom, aoi) }
): Future[Seq[TileLayerCollection[SpatialKey]]] =
Future.sequence {
rasterIds.map { str => Future(cropSingleRasterToAOI(str, zoom, aoi))}
}

/**
* Given a zoom level & area of interest, transform a raster filename into a
Expand Down