You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@mxnet.apache.org by li...@apache.org on 2017/12/04 18:01:15 UTC
[incubator-mxnet] branch master updated: Remove finalizers from
Scala API (#8887)
This is an automated email from the ASF dual-hosted git repository.
liuyizhi pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/incubator-mxnet.git
The following commit(s) were added to refs/heads/master by this push:
new 14cd740 Remove finalizers from Scala API (#8887)
14cd740 is described below
commit 14cd740fbdf52a83a7066f268c5e28d583459666
Author: Calum Leslie <ca...@gmail.com>
AuthorDate: Mon Dec 4 18:01:11 2017 +0000
Remove finalizers from Scala API (#8887)
* [scala] Remove finalizers for leakable resources
Finalizers always run in a separate thread which is not controlled by the user.
Since MXNet cannot be safely accessed from multiple threads, this causes memory
corruption. It is not safe to use finalizers in this way.
This change also adds some leaked object tracing, to allow leaks to be tracked.
It's easy to leak objects and despite warnings in the documentation leaks are
common (even within the Scala API itself). The leak tracing is activated by a
system property, as its runtime cost may be significant.
With the system property unset, the *first* leak of a type will be reported
(without a trace) as a prompt to the developer to investigate.
Co-authored-by: Andre Tamm <ta...@amazon.com>
* [scala] Fix various resource leaks
These leaks were diagnosed with the leak detection added in the previous
commit. This is not an exhaustive clean up but it allows predicting with a
model from Scala at scale (hundreds of millions of comparisons) without a
reported leak, as well as removing the most common errors when training using
the Module code.
Co-authored-by: Andre Tamm <ta...@amazon.com>
---
CONTRIBUTORS.md | 2 +
.../src/main/scala/ml/dmlc/mxnet/Executor.scala | 12 +--
.../src/main/scala/ml/dmlc/mxnet/KVStore.scala | 12 +--
.../src/main/scala/ml/dmlc/mxnet/NDArray.scala | 12 +--
.../core/src/main/scala/ml/dmlc/mxnet/Symbol.scala | 10 +--
.../main/scala/ml/dmlc/mxnet/io/MXDataIter.scala | 12 ++-
.../main/scala/ml/dmlc/mxnet/io/NDArrayIter.scala | 8 +-
.../scala/ml/dmlc/mxnet/module/BaseModule.scala | 19 ++++-
.../mxnet/module/DataParallelExecutorGroup.scala | 16 +++-
.../ml/dmlc/mxnet/util/WarnIfNotDisposed.scala | 88 ++++++++++++++++++++++
.../ml/dmlc/mxnet/util/WarnIfNotDiposedSuite.scala | 63 ++++++++++++++++
11 files changed, 205 insertions(+), 49 deletions(-)
diff --git a/CONTRIBUTORS.md b/CONTRIBUTORS.md
index 76934dc..7209b7c 100644
--- a/CONTRIBUTORS.md
+++ b/CONTRIBUTORS.md
@@ -148,3 +148,5 @@ List of Contributors
* [Jean Kossaifi](https://github.com/JeanKossaifi/)
* [Kenta Kubo](https://github.com/kkk669/)
* [Manu Seth](https://github.com/mseth10/)
+* [Calum Leslie](https://github.com/calumleslie)
+* [Andre Tamm](https://github.com/andretamm)
diff --git a/scala-package/core/src/main/scala/ml/dmlc/mxnet/Executor.scala b/scala-package/core/src/main/scala/ml/dmlc/mxnet/Executor.scala
index 3746a1c..e1d78d1 100644
--- a/scala-package/core/src/main/scala/ml/dmlc/mxnet/Executor.scala
+++ b/scala-package/core/src/main/scala/ml/dmlc/mxnet/Executor.scala
@@ -18,6 +18,7 @@
package ml.dmlc.mxnet
import ml.dmlc.mxnet.Base._
+import org.slf4j.{Logger, LoggerFactory}
import scala.collection.mutable.ArrayBuffer
@@ -34,7 +35,6 @@ object Executor {
* Symbolic Executor component of MXNet <br />
* <b>
* WARNING: it is your responsibility to clear this object through dispose().
- * NEVER rely on the GC strategy
* </b>
*
* @author Yizhi Liu
@@ -44,9 +44,8 @@ object Executor {
* @param symbol
* @see Symbol.bind : to create executor
*/
-// scalastyle:off finalize
class Executor private[mxnet](private[mxnet] val handle: ExecutorHandle,
- private[mxnet] val symbol: Symbol) {
+ private[mxnet] val symbol: Symbol) extends WarnIfNotDisposed {
private[mxnet] var argArrays: Array[NDArray] = null
private[mxnet] var gradArrays: Array[NDArray] = null
private[mxnet] var auxArrays: Array[NDArray] = null
@@ -58,12 +57,10 @@ class Executor private[mxnet](private[mxnet] val handle: ExecutorHandle,
private[mxnet] var _ctx: Context = null
private[mxnet] var _gradsReq: Iterable[_] = null
private[mxnet] var _group2ctx: Map[String, Context] = null
+ private val logger: Logger = LoggerFactory.getLogger(classOf[Executor])
private var disposed = false
-
- override protected def finalize(): Unit = {
- dispose()
- }
+ protected def isDisposed = disposed
def dispose(): Unit = {
if (!disposed) {
@@ -306,4 +303,3 @@ class Executor private[mxnet](private[mxnet] val handle: ExecutorHandle,
str.value
}
}
-// scalastyle:on finalize
diff --git a/scala-package/core/src/main/scala/ml/dmlc/mxnet/KVStore.scala b/scala-package/core/src/main/scala/ml/dmlc/mxnet/KVStore.scala
index 94dd254..3ff606d 100644
--- a/scala-package/core/src/main/scala/ml/dmlc/mxnet/KVStore.scala
+++ b/scala-package/core/src/main/scala/ml/dmlc/mxnet/KVStore.scala
@@ -20,7 +20,7 @@ package ml.dmlc.mxnet
import java.io._
import ml.dmlc.mxnet.Base._
-import org.slf4j.{LoggerFactory, Logger}
+import org.slf4j.{Logger, LoggerFactory}
/**
* Key value store interface of MXNet for parameter synchronization.
@@ -37,7 +37,6 @@ object KVStore {
* Create a new KVStore. <br />
* <b>
* WARNING: it is your responsibility to clear this object through dispose().
- * NEVER rely on the GC strategy
* </b>
*
* @param name : {'local', 'dist'}
@@ -53,15 +52,11 @@ object KVStore {
}
}
-// scalastyle:off finalize
-class KVStore(private[mxnet] val handle: KVStoreHandle) {
+class KVStore(private[mxnet] val handle: KVStoreHandle) extends WarnIfNotDisposed {
private val logger: Logger = LoggerFactory.getLogger(classOf[KVStore])
private var updaterFunc: MXKVStoreUpdater = null
private var disposed = false
-
- override protected def finalize(): Unit = {
- dispose()
- }
+ protected def isDisposed = disposed
/**
* Release the native memory.
@@ -306,4 +301,3 @@ class KVStore(private[mxnet] val handle: KVStoreHandle) {
}
}
}
-// scalastyle:off finalize
diff --git a/scala-package/core/src/main/scala/ml/dmlc/mxnet/NDArray.scala b/scala-package/core/src/main/scala/ml/dmlc/mxnet/NDArray.scala
index 5314dc4..7cfc059 100644
--- a/scala-package/core/src/main/scala/ml/dmlc/mxnet/NDArray.scala
+++ b/scala-package/core/src/main/scala/ml/dmlc/mxnet/NDArray.scala
@@ -17,7 +17,7 @@
package ml.dmlc.mxnet
-import java.nio.{ByteOrder, ByteBuffer}
+import java.nio.{ByteBuffer, ByteOrder}
import ml.dmlc.mxnet.Base._
import ml.dmlc.mxnet.DType.DType
@@ -445,7 +445,7 @@ object NDArray {
val end = retShape.toArray
for (arr <- arrays) {
if (axis == 0) {
- ret.slice(idx, idx + arr.shape(0)).set(arr)
+ ret.slice(idx, idx + arr.shape(0)).set(arr).dispose()
} else {
begin(axis) = idx
end(axis) = idx + arr.shape(axis)
@@ -543,20 +543,15 @@ object NDArray {
* NDArray is basic ndarray/Tensor like data structure in mxnet. <br />
* <b>
* WARNING: it is your responsibility to clear this object through dispose().
- * NEVER rely on the GC strategy
* </b>
*/
-// scalastyle:off finalize
class NDArray private[mxnet](private[mxnet] val handle: NDArrayHandle,
- val writable: Boolean = true) {
+ val writable: Boolean = true) extends WarnIfNotDisposed {
// record arrays who construct this array instance
// we use weak reference to prevent gc blocking
private[mxnet] val dependencies = mutable.HashMap.empty[Long, WeakReference[NDArray]]
private var disposed = false
def isDisposed: Boolean = disposed
- override protected def finalize(): Unit = {
- dispose()
- }
def serialize(): Array[Byte] = {
val buf = ArrayBuffer.empty[Byte]
@@ -1020,7 +1015,6 @@ class NDArray private[mxnet](private[mxnet] val handle: NDArrayHandle,
shape.hashCode + toArray.hashCode
}
}
-// scalastyle:on finalize
private[mxnet] object NDArrayConversions {
implicit def int2Scalar(x: Int): NDArrayConversions = new NDArrayConversions(x.toFloat)
diff --git a/scala-package/core/src/main/scala/ml/dmlc/mxnet/Symbol.scala b/scala-package/core/src/main/scala/ml/dmlc/mxnet/Symbol.scala
index d8da1c6..44b371f 100644
--- a/scala-package/core/src/main/scala/ml/dmlc/mxnet/Symbol.scala
+++ b/scala-package/core/src/main/scala/ml/dmlc/mxnet/Symbol.scala
@@ -27,17 +27,12 @@ import scala.collection.mutable.{ArrayBuffer, ListBuffer}
* Symbolic configuration API of mxnet. <br />
* <b>
* WARNING: it is your responsibility to clear this object through dispose().
- * NEVER rely on the GC strategy
* </b>
*/
-// scalastyle:off finalize
-class Symbol private(private[mxnet] val handle: SymbolHandle) {
+class Symbol private(private[mxnet] val handle: SymbolHandle) extends WarnIfNotDisposed {
private val logger: Logger = LoggerFactory.getLogger(classOf[Symbol])
private var disposed = false
-
- override protected def finalize(): Unit = {
- dispose()
- }
+ protected def isDisposed = disposed
/**
* Release the native memory.
@@ -828,7 +823,6 @@ class Symbol private(private[mxnet] val handle: SymbolHandle) {
}
}
-// scalastyle:on finalize
@AddSymbolFunctions(false)
object Symbol {
private type SymbolCreateNamedFunc = Map[String, Any] => Symbol
diff --git a/scala-package/core/src/main/scala/ml/dmlc/mxnet/io/MXDataIter.scala b/scala-package/core/src/main/scala/ml/dmlc/mxnet/io/MXDataIter.scala
index f964772..32ba359 100644
--- a/scala-package/core/src/main/scala/ml/dmlc/mxnet/io/MXDataIter.scala
+++ b/scala-package/core/src/main/scala/ml/dmlc/mxnet/io/MXDataIter.scala
@@ -18,7 +18,7 @@
package ml.dmlc.mxnet.io
import ml.dmlc.mxnet.Base._
-import ml.dmlc.mxnet.{DataPack, DataBatch, DataIter, NDArray, Shape}
+import ml.dmlc.mxnet.{DataBatch, DataIter, DataPack, NDArray, Shape, WarnIfNotDisposed}
import ml.dmlc.mxnet.IO._
import org.slf4j.LoggerFactory
@@ -29,10 +29,11 @@ import scala.collection.mutable.ListBuffer
* DataIter built in MXNet.
* @param handle the handle to the underlying C++ Data Iterator
*/
-// scalastyle:off finalize
private[mxnet] class MXDataIter(private[mxnet] val handle: DataIterHandle,
dataName: String = "data",
- labelName: String = "label") extends DataIter {
+ labelName: String = "label")
+ extends DataIter with WarnIfNotDisposed {
+
private val logger = LoggerFactory.getLogger(classOf[MXDataIter])
// use currentBatch to implement hasNext
@@ -57,9 +58,7 @@ private[mxnet] class MXDataIter(private[mxnet] val handle: DataIterHandle,
}
private var disposed = false
- override protected def finalize(): Unit = {
- dispose()
- }
+ protected def isDisposed = disposed
/**
* Release the native memory.
@@ -169,7 +168,6 @@ private[mxnet] class MXDataIter(private[mxnet] val handle: DataIterHandle,
override def batchSize: Int = _batchSize
}
-// scalastyle:on finalize
private[mxnet] class MXDataPack(iterName: String, params: Map[String, String]) extends DataPack {
/**
* get data iterator
diff --git a/scala-package/core/src/main/scala/ml/dmlc/mxnet/io/NDArrayIter.scala b/scala-package/core/src/main/scala/ml/dmlc/mxnet/io/NDArrayIter.scala
index e9cb86b..e7dd51b 100644
--- a/scala-package/core/src/main/scala/ml/dmlc/mxnet/io/NDArrayIter.scala
+++ b/scala-package/core/src/main/scala/ml/dmlc/mxnet/io/NDArrayIter.scala
@@ -145,8 +145,12 @@ class NDArrayIter (data: IndexedSeq[NDArray], label: IndexedSeq[NDArray] = Index
private def _padData(ndArray: NDArray): NDArray = {
val padNum = cursor + dataBatchSize - numData
val newArray = NDArray.zeros(ndArray.slice(0, dataBatchSize).shape)
- newArray.slice(0, dataBatchSize - padNum).set(ndArray.slice(cursor, numData))
- newArray.slice(dataBatchSize - padNum, dataBatchSize).set(ndArray.slice(0, padNum))
+ val batch = ndArray.slice(cursor, numData)
+ val padding = ndArray.slice(0, padNum)
+ newArray.slice(0, dataBatchSize - padNum).set(batch).dispose()
+ newArray.slice(dataBatchSize - padNum, dataBatchSize).set(padding).dispose()
+ batch.dispose()
+ padding.dispose()
newArray
}
diff --git a/scala-package/core/src/main/scala/ml/dmlc/mxnet/module/BaseModule.scala b/scala-package/core/src/main/scala/ml/dmlc/mxnet/module/BaseModule.scala
index f6f2e83..1878b35 100644
--- a/scala-package/core/src/main/scala/ml/dmlc/mxnet/module/BaseModule.scala
+++ b/scala-package/core/src/main/scala/ml/dmlc/mxnet/module/BaseModule.scala
@@ -188,6 +188,9 @@ abstract class BaseModule {
batchEndCallback.foreach(callback => {
callback.invoke(epoch, nBatch, evalMetric)
})
+
+ evalBatch.dispose()
+
nBatch += 1
}
@@ -221,6 +224,7 @@ abstract class BaseModule {
while (evalData.hasNext && nBatch != numBatch) {
val evalBatch = evalData.next()
outputList.append(predict(evalBatch))
+ evalBatch.dispose()
nBatch += 1
}
@@ -231,9 +235,12 @@ abstract class BaseModule {
require(binded && paramsInitialized)
forward(batch, isTrain = Option(false))
val pad = batch.pad
- getOutputsMerged().map(out =>
- out.slice(0, out.shape(0)-pad).copy()
- )
+ getOutputsMerged().map(out => {
+ val withoutPadding = out.slice(0, out.shape(0)-pad)
+ val copied = withoutPadding.copy()
+ withoutPadding.dispose()
+ copied
+ })
}
/**
@@ -254,7 +261,9 @@ abstract class BaseModule {
"Cannot merge batches, as num of outputs is not the same in mini-batches." +
"Maybe bucketing is used?")
)
- outputBatches.map(out => NDArray.concatenate(out))
+ val concatenatedOutput = outputBatches.map(out => NDArray.concatenate(out))
+ outputBatches.foreach(_.foreach(_.dispose()))
+ concatenatedOutput
}
// Symbol information
@@ -415,6 +424,8 @@ abstract class BaseModule {
callback.invoke(epoch, nBatch, fitParams.evalMetric)
)
+ dataBatch.dispose()
+
nBatch += 1
}
diff --git a/scala-package/core/src/main/scala/ml/dmlc/mxnet/module/DataParallelExecutorGroup.scala b/scala-package/core/src/main/scala/ml/dmlc/mxnet/module/DataParallelExecutorGroup.scala
index ea78962..010bb1c 100644
--- a/scala-package/core/src/main/scala/ml/dmlc/mxnet/module/DataParallelExecutorGroup.scala
+++ b/scala-package/core/src/main/scala/ml/dmlc/mxnet/module/DataParallelExecutorGroup.scala
@@ -312,8 +312,12 @@ class DataParallelExecutorGroup private[module](
if (labelShapes != None) decideSlices(labelShapes.get)
else null
- private val outputLayouts = symbol.listOutputs().map(name =>
- DataDesc.getBatchAxis(symbol.get(name).attr("__layout__"))
+ private val outputLayouts = symbol.listOutputs().map(name => {
+ val sym = symbol.get(name)
+ val layout = sym.attr("__layout__")
+ sym.dispose()
+ DataDesc.getBatchAxis(layout)
+ }
)
bindExec(dataShapes, labelShapes, sharedGroup)
@@ -599,7 +603,15 @@ class DataParallelExecutorGroup private[module](
label
}
}
+
evalMetric.update(labelsSlice, texec.outputs)
+
+ // Clear up any slices we created (sometimes we don't slice so check for this)
+ (labels zip labelsSlice).foreach { case (label, labelSlice) =>
+ if (label ne labelSlice) {
+ labelSlice.dispose()
+ }
+ }
}
}
diff --git a/scala-package/core/src/main/scala/ml/dmlc/mxnet/util/WarnIfNotDisposed.scala b/scala-package/core/src/main/scala/ml/dmlc/mxnet/util/WarnIfNotDisposed.scala
new file mode 100644
index 0000000..e6fd415
--- /dev/null
+++ b/scala-package/core/src/main/scala/ml/dmlc/mxnet/util/WarnIfNotDisposed.scala
@@ -0,0 +1,88 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package ml.dmlc.mxnet
+
+import org.slf4j.{Logger, LoggerFactory}
+import scala.util.Try
+import scala.collection._
+
+private object WarnIfNotDisposed {
+ private val traceProperty = "mxnet.traceLeakedObjects"
+
+ private val logger: Logger = LoggerFactory.getLogger(classOf[WarnIfNotDisposed])
+
+ // This set represents the list of classes we've logged a warning about if we're not running
+ // in tracing mode. This is used to ensure we only log once.
+ // Don't need to synchronize on this set as it's usually used from a single finalizer thread.
+ private val classesWarned = mutable.Set.empty[String]
+
+ lazy val tracingEnabled = {
+ val value = Try(System.getProperty(traceProperty).toBoolean).getOrElse(false)
+ if (value) {
+ logger.info("Leaked object tracing is enabled (property {} is set)", traceProperty)
+ }
+ value
+ }
+}
+
+// scalastyle:off finalize
+protected trait WarnIfNotDisposed {
+ import WarnIfNotDisposed.logger
+ import WarnIfNotDisposed.traceProperty
+ import WarnIfNotDisposed.classesWarned
+
+ protected def isDisposed: Boolean
+
+ protected val creationTrace: Option[Array[StackTraceElement]] = if (tracingEnabled) {
+ Some(Thread.currentThread().getStackTrace())
+ } else {
+ None
+ }
+
+ override protected def finalize(): Unit = {
+ if (!isDisposed) logDisposeWarning()
+
+ super.finalize()
+ }
+
+ // overridable for testing
+ protected def tracingEnabled = WarnIfNotDisposed.tracingEnabled
+
+ protected def logDisposeWarning(): Unit = {
+ // The ":Any" casts below are working around the Slf4j Java API having overloaded methods that
+ // Scala doesn't resolve automatically.
+ if (creationTrace.isDefined) {
+ logger.warn(
+ "LEAK: An instance of {} was not disposed. Creation point of this resource was:\n\t{}",
+ getClass(), creationTrace.get.mkString("\n\t"): Any)
+ } else {
+ // Tracing disabled but we still warn the first time we see a leak to ensure the code author
+ // knows. We could warn every time but this can be very noisy.
+ val className = getClass().getName()
+ if (!classesWarned.contains(className)) {
+ logger.warn(
+ "LEAK: [one-time warning] An instance of {} was not disposed. " + //
+ "Set property {} to true to enable tracing",
+ className, traceProperty: Any)
+
+ classesWarned += className
+ }
+ }
+ }
+}
+// scalastyle:on finalize
diff --git a/scala-package/core/src/test/scala/ml/dmlc/mxnet/util/WarnIfNotDiposedSuite.scala b/scala-package/core/src/test/scala/ml/dmlc/mxnet/util/WarnIfNotDiposedSuite.scala
new file mode 100644
index 0000000..ffd9a69
--- /dev/null
+++ b/scala-package/core/src/test/scala/ml/dmlc/mxnet/util/WarnIfNotDiposedSuite.scala
@@ -0,0 +1,63 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package ml.dmlc.mxnet
+
+import org.scalatest.{BeforeAndAfterAll, FunSuite}
+
+// scalastyle:off finalize
+class Leakable(enableTracing: Boolean = false, markDisposed: Boolean = false)
+ extends WarnIfNotDisposed {
+ def isDisposed: Boolean = markDisposed
+ override protected def tracingEnabled = enableTracing
+
+ var warningWasLogged: Boolean = false
+ def getCreationTrace: Option[Array[StackTraceElement]] = creationTrace
+
+ override def finalize(): Unit = super.finalize()
+ override protected def logDisposeWarning() = {
+ warningWasLogged = true
+ }
+}
+// scalastyle:on finalize
+
+class WarnIfNotDisposedSuite extends FunSuite with BeforeAndAfterAll {
+ test("trace collected if tracing enabled") {
+ val leakable = new Leakable(enableTracing = true)
+
+ val trace = leakable.getCreationTrace
+ assert(trace.isDefined)
+ assert(trace.get.exists(el => el.getClassName() == getClass().getName()))
+ }
+
+ test("trace not collected if tracing disabled") {
+ val leakable = new Leakable(enableTracing = false)
+ assert(!leakable.getCreationTrace.isDefined)
+ }
+
+ test("no warning logged if object disposed") {
+ val notLeaked = new Leakable(markDisposed = true)
+ notLeaked.finalize()
+ assert(!notLeaked.warningWasLogged)
+ }
+
+ test("warning logged if object not disposed") {
+ val leaked = new Leakable(markDisposed = false)
+ leaked.finalize()
+ assert(leaked.warningWasLogged)
+ }
+}
--
To stop receiving notification emails like this one, please contact
['"commits@mxnet.apache.org" <co...@mxnet.apache.org>'].