MNEMONIC-259: Add test cases to verify its functionality of the DurableRDD Direct IO

Author: Wang, Gang(Gary)

build-tools/test.conf

@@ -69,3 +69,5 @@ mvn -Dsuites=org.apache.mnemonic.spark.rdd.DurableRDDPersonDataSpec test -pl mne
 
 mvn -Dsuites=org.apache.mnemonic.spark.rdd.DurableRDDBufferDataSpec test -pl mnemonic-spark/mnemonic-spark-core -DskipTests=false
 
+mvn -Dsuites=org.apache.mnemonic.spark.DurableBufferIODataSpec test -pl mnemonic-spark/mnemonic-spark-core -DskipTests=false
+

mnemonic-spark/mnemonic-spark-core/src/main/scala/org/apache/mnemonic/spark/DurableSparkFunctions.scala

@@ -37,5 +37,5 @@ class DurableSparkFunctions(sc: SparkContext) extends Serializable {
 }
 
 object DurableSparkFunctions {
-  implicit def addDurableFunctions(sc: SparkContext) = new DurableSparkFunctions(sc)
+  implicit def addDurableSparkFunctions(sc: SparkContext) = new DurableSparkFunctions(sc)
 }

mnemonic-spark/mnemonic-spark-core/src/main/scala/org/apache/mnemonic/spark/rdd/DurableRDD.scala

@@ -44,7 +44,7 @@ private[spark] class DurableRDD[D: ClassTag, T: ClassTag] (
   preservesPartitioning: Boolean = false)
     extends RDD[D](_sc, deps) {
 
-  private val isInputOnly =  null == deps
+  private val isInputOnly =  Nil == deps
 
   private var durdddir:String = _
   if (isInputOnly) {
@@ -64,7 +64,9 @@ private[spark] class DurableRDD[D: ClassTag, T: ClassTag] (
     if (isInputOnly) {
       val ret = DurableRDD.collectMemPoolPartitionList(durdddir).getOrElse(Array[Partition]())
       if (ret.isEmpty) {
-        logInfo(s"Not found any partitions in the directory ${durdddir}")
+        logWarning(s"Not found any partitions in the directory ${durdddir}")
+      } else {
+        logInfo(s"Found ${ret.length} partitions in the directory ${durdddir}")
       }
       ret
     } else {
@@ -78,7 +80,7 @@ private[spark] class DurableRDD[D: ClassTag, T: ClassTag] (
     val memplist = mempListOpt match {
       case None => {
         if (isInputOnly) {
-          logInfo(s"Not found any mem pool files related to the partition #${context.partitionId}")
+          logWarning(s"Not found any mem pool files related to the partition #${context.partitionId}")
           Array[File]()
         } else {
           val mplst = DurableRDD.prepareDurablePartition[D, T](durdddir,
@@ -112,7 +114,7 @@ object DurableRDD {
 
   val durableSubDirNameTemplate = "durable-rdd-%010d"
   val durableFileNameTemplate = "mem_%010d_%010d.mne"
-  val durableFileNamePartitionRegex = raw"mem_(\d{10})_0000000000.mne".r
+  val durableFileNamePartitionRegex = raw".*mem_(\d{10})_0000000000.mne".r
 
   private var durableDir: Option[String] = None
 
@@ -176,7 +178,7 @@ object DurableRDD {
     val paridset = new mutable.TreeSet[Int]
     val dir = new File(path)
     if (dir.isDirectory) {
-      val flst = dir.listFiles.filter(_.isDirectory)
+      val flst = dir.listFiles.filter(!_.isDirectory)
       for (file <- flst) {
         file.toString match {
           case durableFileNamePartitionRegex(paridx) => {
@@ -186,7 +188,7 @@ object DurableRDD {
         }
       }
     }
-    Option(paridset.toArray.map(x => new Partition { val index = x }))
+    Option(paridset.toArray.map(x => new Partition with Serializable { override def index:Int = x }))
   }
 
   def collectMemPoolFileList(durddir: String, memFileNameGen: (Long)=>String): Option[Array[File]] = {
@@ -253,7 +255,7 @@ object DurableRDD {
       sc: SparkContext, path: String,
       serviceName: String, durableTypes: Array[DurableType],
       entityFactoryProxies: Array[EntityFactoryProxy], slotKeyId: Long) = {
-    val ret = new DurableRDD[D, Unit](sc, null,
+    val ret = new DurableRDD[D, Unit](sc, Nil,
       serviceName, durableTypes, entityFactoryProxies, slotKeyId,
       1024*1024*1024L, path, null)
     ret

mnemonic-spark/mnemonic-spark-core/src/main/scala/org/apache/mnemonic/spark/rdd/DurableRDDFunctions.scala

@@ -25,6 +25,7 @@ import org.apache.mnemonic.NonVolatileMemAllocator
 import org.apache.mnemonic.DurableType
 import org.apache.mnemonic.EntityFactoryProxy
 import org.apache.mnemonic.sessions.ObjectCreator
+import org.apache.commons.io.FileUtils
 
 class DurableRDDFunctions[T: ClassTag](rdd: RDD[T]) extends Serializable {
 
@@ -41,14 +42,17 @@ class DurableRDDFunctions[T: ClassTag](rdd: RDD[T]) extends Serializable {
       partitionPoolSize, f, preservesPartitioning)
   }
 
-  def saveAsMnemonic[D: ClassTag] (path: String,
+  def saveAsMnemonic[D: ClassTag] (path: String, cleanPath: Boolean,
                      serviceName: String,
                      durableTypes: Array[DurableType],
                      entityFactoryProxies: Array[EntityFactoryProxy],
                      slotKeyId: Long,
                      partitionPoolSize: Long,
                      f: (T, ObjectCreator[D, NonVolatileMemAllocator]) => Option[D]) {
     val dir = new File(path)
+    if (cleanPath && dir.exists) {
+      FileUtils.deleteDirectory(dir)
+    }
     if (!dir.exists) {
       dir.mkdir
     }
@@ -61,5 +65,5 @@ class DurableRDDFunctions[T: ClassTag](rdd: RDD[T]) extends Serializable {
 }
 
 object DurableRDDFunctions {
-  implicit def addDurableFunctions[T: ClassTag](rdd: RDD[T]) = new DurableRDDFunctions[T](rdd)
+  implicit def addDurableRDDFunctions[T: ClassTag](rdd: RDD[T]) = new DurableRDDFunctions[T](rdd)
 }

mnemonic-spark/mnemonic-spark-core/src/test/scala/org/apache/mnemonic/spark/DurableBufferIODataSpec.scala

@@ -0,0 +1,101 @@
+/*
+ * 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 org.apache.mnemonic.spark
+
+import java.util.zip.CRC32
+
+import org.apache.mnemonic.{DurableBuffer, DurableType}
+import org.apache.mnemonic.sessions.ObjectCreator
+import org.apache.mnemonic.spark.rdd.DurableRDDFunctions._
+import org.apache.mnemonic.spark.DurableSparkFunctions._
+
+import org.apache.spark._
+import org.apache.spark.rdd._
+
+import scala.language.existentials
+import scala.util._
+
+class DurableBufferIODataSpec extends TestSpec {
+
+  val defaultServiceName = "pmalloc"
+  val defaultSlotKeyId = 2L
+  val defaultPartitionSize = 1024 * 1024 * 1024L
+  val defaultBaseDirectory = "."
+  val defaultNumOfPartitions = 8
+  val defaultNumOfRecordsPerPartition = 200
+  val mneIODirPath = "./target/mne_io_data"
+
+  behavior of "A SparkContext with DurableRDD"
+
+  it should "supports durable buffer for direct IO operations" in {
+    val dataOffset = 8
+    val conf = new SparkConf()
+      .setMaster("local[*]")
+      .setAppName("Test")
+    val sc = new SparkContext(conf)
+    val seed: RDD[Int] = sc.parallelize(
+      Seq.fill(defaultNumOfPartitions)(defaultNumOfRecordsPerPartition), defaultNumOfPartitions)
+    val data = seed flatMap (recnum => Seq.fill(recnum)(Random.nextInt(1024 * 1024) + 1024 * 1024)) cache
+
+    val durdd = data.saveAsMnemonic[DurableBuffer[_]](
+      mneIODirPath, true,
+      defaultServiceName,
+      Array(DurableType.BUFFER), Array(),
+      defaultSlotKeyId, defaultPartitionSize,
+      (v: Int, oc: ObjectCreator[DurableBuffer[_], _])=>
+      {
+        val cs = new CRC32
+        cs.reset
+        val buffer = oc.newDurableObjectRecord(v)
+        val bary = new Array[Byte](v - dataOffset)
+        if (null != buffer) {
+          buffer.clear
+          Random.nextBytes(bary)
+          cs.update(bary, 0, bary.length)
+          buffer.get.putLong(cs.getValue)
+          buffer.get.put(bary)
+        }
+        Option(buffer)
+      })
+
+    val indurdd = sc.mnemonic[DurableBuffer[_]](mneIODirPath, defaultServiceName,
+      Array(DurableType.BUFFER), Array(), defaultSlotKeyId)
+    val durtsz = indurdd map (_.getSize.toInt) sum
+    val derrcount = indurdd map (
+      buffer => {
+        var chksum: Long = -1L
+        val cs = new CRC32
+        cs.reset
+        if (null != buffer) {
+          buffer.clear
+          chksum = buffer.get.getLong
+          val bary = new Array[Byte](buffer.get.remaining)
+          buffer.get.get(bary)
+          cs.update(bary)
+        }
+        if (chksum != cs.getValue) 1 else 0
+      }) sum
+
+    val (rerrcnt: Long, rsz: Long) = (0L, data.sum.toLong)
+    val (derrcnt: Long, dsz: Long) = (derrcount.toLong, durtsz.toLong)
+
+    assertResult((rerrcnt, rsz)) {
+      (derrcnt, dsz)
+    }
+  }
+}